resolve: 64 MiB redeem ceiling (a comic chapter redeems as one CBZ)

.gitea/workflows/publish.yml

@@ -0,0 +1,86 @@
+# Gitea twin of .github/workflows/publish.yml. Gitea runs only .gitea/workflows/ once this
+# folder exists, and GitHub never reads it.
+#
+# Publishes to IPCR on watch.nsl.sh instead of ghcr.io:
+#   push  ipcr-push-85-17-246-67.nip.io/metadec.eth/<repo>:<tag>   docker login as metadec.eth with the
+#         IPCR_PUSH_TOKEN secret (the metadec.eth namespace token from the IPCR dashboard)
+#   pull  ipcr-watch.nsl.sh/ipns/metadec.eth/metadec.eth/<repo>:<tag>   anonymous
+# The push goes to ipcr-push-85-17-246-67.nip.io, the box's own address, not ipcr-push-watch.nsl.sh: Cloudflare fronts
+# *.watch.nsl.sh and refuses any request over 100 MB (413), which a large image layer is.
+# That host's certificate is signed by the box's private Mesh Router CA, given to buildkit from
+# the IPCR_CA_PEM repo variable. docker login would make the job's Docker daemon check the
+# certificate without that CA, so the login is written straight into ~/.docker/config.json.
+# The runner's job image (node:22-bookworm) has no docker CLI, so the static one is
+# installed first; buildx installs itself. No `type=gha` cache: the runner has none.
+
+name: Publish image
+
+# Build and push the Docker image to the IPCR
+# on main (tags main + short sha) and on v* tags (version; latest).
+on:
+  push:
+    branches:
+      - main
+    tags:
+      - 'v*'
+
+env:
+  REGISTRY: ipcr-push-85-17-246-67.nip.io
+  IMAGE_NAME: metadec.eth/${{ github.event.repository.name }}
+
+jobs:
+  publish:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Install Docker CLI
+        run: curl -fsSL https://download.docker.com/linux/static/stable/x86_64/docker-28.5.1.tgz | tar -xz -C /usr/local/bin --strip-components=1 docker/docker
+
+      - name: Trust the Mesh Router CA
+        run: printf '%s\n' "$IPCR_CA_PEM" > /tmp/ipcr-ca.pem
+        env:
+          IPCR_CA_PEM: ${{ vars.IPCR_CA_PEM }}
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+        with:
+          version: latest
+          buildkitd-config-inline: |
+            [registry."${{ env.REGISTRY }}"]
+              ca = ["/tmp/ipcr-ca.pem"]
+
+      - name: Log in to IPCR
+        env:
+          IPCR_PUSH_TOKEN: ${{ secrets.IPCR_PUSH_TOKEN }}
+        run: |
+          mkdir -p ~/.docker
+          printf '{"auths":{"%s":{"auth":"%s"}}}\n' "$REGISTRY" "$(printf 'metadec.eth:%s' "$IPCR_PUSH_TOKEN" | base64 -w0)" > ~/.docker/config.json
+
+      - name: Extract image metadata
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
+          # v1.2.3 -> 1.2.3, 1.2, 1 ; plus a moving `latest` for the newest tag.
+          tags: |
+            type=semver,pattern={{version}}
+            type=semver,pattern={{major}}.{{minor}}
+            type=semver,pattern={{major}}
+            type=ref,event=branch
+            type=sha,format=short
+            # latest = newest release, not main
+            type=raw,value=latest,enable=${{ startsWith(github.ref, 'refs/tags/v') }}
+
+      - name: Build and push
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          file: ./Dockerfile
+          push: true
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}

CLAUDE.md

@@ -7,6 +7,11 @@ Behaviour must stay identical to the monolith's swarm / blockstore / ingress.
 
 1. **The swarm is transport-only**: identify/kad/mdns/bitswap. No kamilata, no
    gossipsub, no request_response — search is meta-search's.
+   Resolving a **pointer** (`GET /ipfs-tier/resolve/:cid`, `resolve.rs`) is not a
+   swarm protocol: it redeems over plain HTTP at a gateway's base URL, chosen by
+   `gateways.rs` from the config-pinned local gateways plus the swarm directory,
+   fastest first. Never move it onto libp2p, and never let bitswap answer a
+   pointer — its multihash is `identity`, there is nothing to verify.
 2. **`MAX_MULTIHASH_SIZE = 64`** must match meta-gateway and meta-search.
 3. **`core2` is patched** in `Cargo.toml` (every libp2p-0.56 bitswap impl needs
    the yanked `^0.4.0`); mirror changes in meta-gateway.

Cargo.lock

@@ -1964,7 +1964,8 @@ checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98"
 
 [[package]]
 name = "meta-feeder-sdk"
-version = "1.4.0"
+version = "1.8.0"
+source = "git+https://github.com/worph/meta-feeder-sdk?tag=v1.8.0#612f2aed661925cd22d51b29564832991e246e4d"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -1989,7 +1990,7 @@ dependencies = [
 
 [[package]]
 name = "meta-transport-ipfs"
-version = "1.0.0"
+version = "1.3.1"
 dependencies = [
  "anyhow",
  "async-trait",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "meta-transport-ipfs"
-version = "1.0.0"
+version = "1.3.1"
 edition = "2021"
 rust-version = "1.85"
 license = "MIT"
@@ -19,7 +19,7 @@ path = "src/main.rs"
 # The transport-plugin contract. Pinned by TAG, never by path — a path dep
 # resolves outside this repo, which is the Docker build context. See the
 # Dockerfile for building against an unreleased SDK.
-meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.4.0", features = ["transport"] }
+meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.8.0", features = ["transport"] }
 
 # libp2p — transport only: identify (address exchange, load-bearing for bitswap
 # reachability), kad (DHT provider records), mdns (LAN discovery) + bitswap.
@@ -56,7 +56,7 @@ multihash-codetable = { version = "0.1", features = ["sha2", "sha3"] }
 
 [dev-dependencies]
 tempfile = "3"
-meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.4.0", features = ["testkit"] }
+meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.8.0", features = ["testkit"] }
 
 # ---------------------------------------------------------------------------
 # Workaround for the yanked transitive `core2` (all versions). `multihash-

Dockerfile

@@ -44,8 +44,8 @@ COPY --from=builder /meta-transport-ipfs /usr/local/bin/meta-transport-ipfs
 
 # Same uid as meta-share: the plugins share its /data volume.
 RUN useradd --system --uid 10001 --no-create-home metashare \
-    && mkdir -p /data/meta-share \
-    && chown metashare:metashare /data/meta-share
+    && mkdir -p /data/meta-share /state \
+    && chown metashare:metashare /data/meta-share /state
 USER metashare
 WORKDIR /data/meta-share
 
@@ -57,7 +57,7 @@ EXPOSE 3000 4001
 ENV HTTP_LISTEN=0.0.0.0:3000 \
     P2P_LISTEN=/ip4/0.0.0.0/tcp/4001 \
     META_SHARE_DATA=/data/meta-share \
-    META_SHARE_CONFIG_DIR=/config \
+    META_SHARE_PLUGIN_STATE_DIR=/state \
     RUST_LOG=info,meta_transport_ipfs=info,libp2p=warn,beetswap=warn,yamux=warn
 
 ENTRYPOINT ["/usr/local/bin/meta-transport-ipfs"]

README.md

@@ -23,11 +23,20 @@ meta-share's variables, unchanged: `P2P_LISTEN`, `PUBLIC_ADDR`, `KAD_*`,
 `META_SHARE_BLOCKSTORE_CACHE_MB`, `META_SHARE_INGRESS_*`,
 `META_SHARE_LOOSE_BLOCK_MAX_BYTES`, `META_SHARE_LOCAL_FILES_PATH`,
 `META_SHARE_FILES_PATH`, `META_SHARE_SEED_DHT_PROVIDE`, `META_SHARE_DATA`, plus
-`META_SHARE_HULL_URL`. `meta_core_url` / `peer_url` come from the hull's
-`settings.json` (mount `/config` read-only), env as the first-boot fallback.
+`META_SHARE_HULL_URL`. `meta_core_url` / `peer_url` are asked from the hull at
+boot (`GET /internal/network`), env as the fallback.
 
-Start it **after** the hull is healthy: the hull migrates its material index out
-of `ipfs/blocks.redb` on its first boot after the split (redb is single-process).
+**Own settings** (config plane, `META_SHARE_PLUGIN_STATE_DIR`, default `/state` →
+`config.json`): cohort namespace, DHT/pinned bootstrap peers, mDNS, DHT announce,
+max connections, background completion. Edited from meta-share's dashboard
+(Plugins → ipfs → configure); `config.json` wins and is written onto the env names
+above at boot, so env only seeds it. Saving restarts the plugin. `PUBLIC_ADDR` /
+`P2P_LISTEN` stay env-only (they follow the published port and the box's IP).
+
+Boot order is handled here, not by compose: before opening `ipfs/blocks.redb` the
+plugin waits (≤ 60 s) for the hull's `/health`, because the hull migrates its
+material index out of that file on its first boot after the split and redb is
+single-process.
 
 ## Build
 

src/api/mod.rs

@@ -38,6 +38,8 @@ pub struct AppState {
     /// default = public mode). Decided here, where `PUBLIC_ADDR` is known; the
     /// hull always asks and this gates.
     pub seed_dht_provide: bool,
+    /// Gateways that can redeem pointers, fastest first (`/ipfs-tier/resolve`).
+    pub gateways: Arc<crate::gateways::Gateways>,
 }
 
 /// "This peer now seeds `cid`" — the hull writes the seed row (and decides on

src/config.rs

@@ -268,7 +268,10 @@ pub struct Config {
 }
 
 impl Config {
-    pub fn from_env() -> Result<Self> {
+    /// `net`: the hull's endpoints (`GET /internal/network`), or
+    /// [`NetworkSettings::from_env`](crate::settings::NetworkSettings::from_env)
+    /// when the hull did not answer.
+    pub fn from_env(net: crate::settings::NetworkSettings) -> Result<Self> {
         // Reachability posture first — the listen addr, the kad bootstrap
         // list, and whether we publish provider records all fall out of it.
         // `P2P_ANNOUNCE` is the deprecated spelling of `PUBLIC_ADDR`.
@@ -323,13 +326,7 @@ impl Config {
             )),
         };
 
-        // The network endpoints are the hull's settings (`settings.json`,
-        // seeded once from env by the hull). Read-only here; env fallback when
-        // the file doesn't exist yet.
-        let config_dir: PathBuf = std::env::var("META_SHARE_CONFIG_DIR")
-            .unwrap_or_else(|_| "/config".to_string())
-            .into();
-        let net = crate::settings::NetworkSettings::load(&config_dir);
+        // The network endpoints are the hull's settings, handed in by `main`.
         let peer_api_url = net
             .peer_url
             .as_deref()

src/gateways.rs

@@ -0,0 +1,205 @@
+//! Which gateways can redeem a pointer, fastest first.
+//!
+//! Two sources, merged by base URL:
+//!
+//! - **Pinned** gateways from config (`gateway_urls`, seeded from
+//!   `META_SHARE_GATEWAY_URLS`) — the local one. Their redeem claims come
+//!   straight from `GET {url}/api/gateway/plugins`, so they never depend on the
+//!   libp2p link (watch.nsl.sh 2026-10-04: a lost swarm link left every NZB
+//!   "unredeemable" though the gateway sat on the same Docker network).
+//! - **Swarm** gateways from the peer directory (identify `gateways=` + a caps
+//!   fetch) — remote ones.
+//!
+//! Order is by the round-trip of the last caps fetch, which the refresh task
+//! times for every known gateway. Nothing ranks the local gateway first by rule:
+//! it is simply the closest, so it wins on latency.
+
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+use tracing::debug;
+
+use crate::api::AppState;
+use crate::gateway_discovery::{fetch_gateway_caps, RedeemClaim};
+use crate::swarm::PeerDirectory;
+
+/// How often pinned gateways' claims and every gateway's latency are refreshed.
+const REFRESH_EVERY: Duration = Duration::from_secs(30);
+/// A pinned gateway's claims are trusted this long after the last fetch.
+const CLAIMS_TTL: Duration = Duration::from_secs(600);
+const FETCH_TIMEOUT: Duration = Duration::from_secs(5);
+
+#[derive(Default)]
+pub struct Gateways {
+    pinned: Vec<String>,
+    /// Pinned base URL → (claims, fetched at).
+    pinned_claims: Mutex<HashMap<String, (Vec<RedeemClaim>, Instant)>>,
+    /// Base URL → last caps-fetch round-trip.
+    rtt: Mutex<HashMap<String, Duration>>,
+}
+
+fn norm(url: &str) -> String {
+    url.trim().trim_end_matches('/').to_string()
+}
+
+impl Gateways {
+    pub fn new(pinned: impl IntoIterator<Item = String>) -> Self {
+        let mut list: Vec<String> = Vec::new();
+        for u in pinned.into_iter().map(|u| norm(&u)).filter(|u| !u.is_empty()) {
+            if !list.contains(&u) {
+                list.push(u);
+            }
+        }
+        Self { pinned: list, ..Self::default() }
+    }
+
+    /// From `META_SHARE_GATEWAY_URLS` (comma-separated), which the config plane
+    /// overlays at boot.
+    pub fn from_env() -> Self {
+        Self::new(
+            std::env::var("META_SHARE_GATEWAY_URLS")
+                .unwrap_or_default()
+                .split(',')
+                .map(str::to_string),
+        )
+    }
+
+    pub fn pinned(&self) -> &[String] {
+        &self.pinned
+    }
+
+    fn record(&self, url: &str, claims: Option<Vec<RedeemClaim>>, rtt: Duration) {
+        self.rtt.lock().unwrap().insert(norm(url), rtt);
+        if let Some(c) = claims {
+            self.pinned_claims.lock().unwrap().insert(norm(url), (c, Instant::now()));
+        }
+    }
+
+    /// Gateways whose claim covers (`codec`, `key`), fastest first. A gateway
+    /// with no measured latency yet goes after every measured one.
+    pub fn candidates(&self, dir: &PeerDirectory, codec: &str, key: &str) -> Vec<String> {
+        let mut out: Vec<String> = Vec::new();
+        {
+            let claims = self.pinned_claims.lock().unwrap();
+            for url in &self.pinned {
+                let covered = claims.get(url).is_some_and(|(c, at)| {
+                    at.elapsed() <= CLAIMS_TTL && c.iter().any(|c| c.covers(codec, key))
+                });
+                if covered {
+                    out.push(url.clone());
+                }
+            }
+        }
+        for url in dir.gateways_redeeming(codec, key) {
+            let url = norm(&url);
+            if !out.contains(&url) {
+                out.push(url);
+            }
+        }
+        self.by_latency(out)
+    }
+
+    fn by_latency(&self, mut urls: Vec<String>) -> Vec<String> {
+        let rtt = self.rtt.lock().unwrap();
+        // Stable sort: unmeasured gateways keep their order (pinned first).
+        urls.sort_by_key(|u| rtt.get(u).copied().unwrap_or(Duration::MAX));
+        urls
+    }
+
+    /// Is any gateway known at all — a pinned one that answered, or a swarm one?
+    /// Tells "nobody claims this" apart from "discovery hasn't finished".
+    pub fn knows_any(&self, dir: &PeerDirectory) -> bool {
+        let pinned_answered = self
+            .pinned_claims
+            .lock()
+            .unwrap()
+            .values()
+            .any(|(_, at)| at.elapsed() <= CLAIMS_TTL);
+        pinned_answered || dir.knows_a_gateway()
+    }
+}
+
+/// Refresh pinned claims and every gateway's latency, now and every
+/// [`REFRESH_EVERY`].
+pub fn spawn_refresh(state: Arc<AppState>) {
+    tokio::spawn(async move {
+        loop {
+            refresh(&state).await;
+            tokio::time::sleep(REFRESH_EVERY).await;
+        }
+    });
+}
+
+async fn refresh(state: &AppState) {
+    let gws = &state.gateways;
+    let mut urls: Vec<(String, bool)> = gws.pinned().iter().map(|u| (u.clone(), true)).collect();
+    for p in state.peer_directory.snapshot() {
+        if p.redeems.is_empty() {
+            continue;
+        }
+        if let Some(u) = p.base_url.as_deref().map(norm) {
+            if !urls.iter().any(|(x, _)| *x == u) {
+                urls.push((u, false));
+            }
+        }
+    }
+    for (url, pinned) in urls {
+        let t0 = Instant::now();
+        match fetch_gateway_caps(&state.http, &url, FETCH_TIMEOUT).await {
+            Ok(caps) => gws.record(&url, pinned.then(|| caps.redeem_claims()), t0.elapsed()),
+            Err(e) => debug!(gateway = %url, error = %e, "gateways: caps fetch failed"),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::gateway_discovery::REDEEM_NZB_RELEASE;
+
+    fn claim(host: &str) -> RedeemClaim {
+        serde_json::from_value(serde_json::json!({
+            "codec": REDEEM_NZB_RELEASE, "field": "manifest", "hosts": [host], "sources": []
+        }))
+        .unwrap()
+    }
+
+    #[test]
+    fn a_pinned_gateway_counts_only_for_what_it_claims() {
+        let g = Gateways::new(["http://local:3000/".to_string()]);
+        let dir = PeerDirectory::new();
+        assert!(g.candidates(&dir, REDEEM_NZB_RELEASE, "api.nzbgeek.info").is_empty());
+        assert!(!g.knows_any(&dir));
+        g.record("http://local:3000", Some(vec![claim("api.nzbgeek.info")]), Duration::from_millis(3));
+        assert_eq!(
+            g.candidates(&dir, REDEEM_NZB_RELEASE, "api.nzbgeek.info"),
+            vec!["http://local:3000".to_string()]
+        );
+        assert!(g.candidates(&dir, REDEEM_NZB_RELEASE, "api.nzb.life").is_empty());
+        assert!(g.knows_any(&dir));
+    }
+
+    #[test]
+    fn the_fastest_gateway_comes_first() {
+        let g = Gateways::new(["http://a".to_string(), "http://b".to_string(), "http://c".to_string()]);
+        g.record("http://a", Some(vec![claim("h")]), Duration::from_millis(80));
+        g.record("http://b", Some(vec![claim("h")]), Duration::from_millis(2));
+        // c claims but was never timed separately from its claims fetch: still timed.
+        g.record("http://c", Some(vec![claim("h")]), Duration::from_millis(20));
+        assert_eq!(
+            g.candidates(&PeerDirectory::new(), REDEEM_NZB_RELEASE, "h"),
+            vec!["http://b", "http://c", "http://a"]
+        );
+    }
+
+    #[test]
+    fn unmeasured_gateways_go_last_in_their_given_order() {
+        let g = Gateways::new([]);
+        g.rtt.lock().unwrap().insert("http://timed".into(), Duration::from_millis(500));
+        assert_eq!(
+            g.by_latency(vec!["http://x".into(), "http://timed".into(), "http://y".into()]),
+            vec!["http://timed", "http://x", "http://y"]
+        );
+    }
+}

src/main.rs

@@ -6,20 +6,26 @@
 //! `cache/`. Serves the transport contract on `HTTP_LISTEN` (internal only) and
 //! the public `/ipfs/:cid` gateway, which the hull relays.
 //!
-//! The swarm announces the **hull's** URL in identify (`baseUrl=`, from the
-//! hull's `settings.json`), so peers and gateways keep calling meta-share's
-//! public API exactly as before — the plugin is invisible on the wire.
+//! The swarm announces the **hull's** URL in identify (`baseUrl=`, asked from the
+//! hull at boot), so peers and gateways keep calling meta-share's public API
+//! exactly as before — the plugin is invisible on the wire.
+//!
+//! Own settings: [`plane`] (`<state dir>/config.json`, edited from meta-share's
+//! dashboard), overlaid onto the env names the modules read.
 
 mod api;
 mod blockstore;
 mod config;
 mod filestore;
 mod gateway_discovery;
+mod gateways;
 mod ingress;
 mod ingress_commit;
 mod ipfs_chunk;
 mod material;
+mod plane;
 mod plugin;
+mod resolve;
 mod settings;
 mod share;
 mod store;
@@ -35,7 +41,8 @@ use std::sync::Arc;
 
 use anyhow::{Context, Result};
 use libp2p::identity;
-use meta_feeder_sdk::transport::{serve_transport, FocusView, HullClient};
+use meta_feeder_sdk::transport::config::state_dir_from_env;
+use meta_feeder_sdk::transport::{serve_transport, ConfigPlane, FocusView, HullClient};
 use tracing::{debug, info, warn};
 use tracing_subscriber::EnvFilter;
 
@@ -46,7 +53,35 @@ async fn main() -> Result<()> {
     });
     tracing_subscriber::fmt().with_env_filter(filter).with_target(false).init();
 
-    let cfg = config::Config::from_env()?;
+    // Own settings first: `config.json` wins over the env it was seeded from,
+    // written onto the env names every module reads — before any of them does.
+    let plane = Arc::new(ConfigPlane::new(plane::schema(), &state_dir_from_env(), plane::seed()));
+    plane::overlay_env(&plane.effective());
+
+    let http = reqwest::Client::builder()
+        .user_agent(concat!("meta-transport-ipfs/", env!("CARGO_PKG_VERSION")))
+        .timeout(std::time::Duration::from_secs(30))
+        .build()
+        .context("build shared reqwest client")?;
+    let hull = HullClient::from_env(http.clone());
+
+    // Wait for the hull before opening `ipfs/blocks.redb`: on its first boot
+    // after the split it migrates the material index out of that file, and redb
+    // is single-process. Compose `depends_on` used to order this; separate
+    // store apps cannot. Bounded — a hull that stays down must not wedge us.
+    if hull.is_enabled() && !hull.wait_ready(std::time::Duration::from_secs(60)).await {
+        warn!("meta-share (hull) did not answer within 60s; starting anyway");
+    }
+    let net = match hull.network().await {
+        Ok(Some(n)) => settings::NetworkSettings::from(n),
+        Ok(None) => settings::NetworkSettings::from_env(),
+        Err(e) => {
+            warn!(error = %e, "could not ask the hull for its endpoints; using env");
+            settings::NetworkSettings::from_env()
+        }
+    };
+
+    let cfg = config::Config::from_env(net)?;
     let keypair = identity::Keypair::generate_ed25519();
     let local_peer_id = libp2p::PeerId::from(keypair.public());
     let agent_version = swarm::build_agent_version(cfg.peer_api_url.as_deref());
@@ -61,17 +96,13 @@ async fn main() -> Result<()> {
         kad_provide_enabled = cfg.kad.provide_enabled,
         seed_dht_provide = cfg.seed_dht_provide,
         agent_version = %agent_version,
+        config = %plane.path().display(),
         "starting meta-transport-ipfs"
     );
 
     let raw_block_store = blockstore::open_redb_blockstore(&cfg.data_dir)
         .await
         .context("open ipfs blockstore")?;
-    let http = reqwest::Client::builder()
-        .user_agent(concat!("meta-transport-ipfs/", env!("CARGO_PKG_VERSION")))
-        .timeout(std::time::Duration::from_secs(30))
-        .build()
-        .context("build shared reqwest client")?;
     let webdav_url_cache = Arc::new(tokio::sync::OnceCell::new());
     let redb_handle = raw_block_store.raw_db();
     let ingress = Arc::new(ingress::IngressRegistry::from_data_dir(&cfg.data_dir));
@@ -132,12 +163,14 @@ async fn main() -> Result<()> {
         peer_directory,
         focus,
         http: http.clone(),
-        hull: HullClient::from_env(http),
+        hull,
         data_dir: cfg.data_dir.clone(),
         meta_core_url: cfg.meta_core_url.clone(),
         webdav_url_cache,
         seed_dht_provide: cfg.seed_dht_provide,
+        gateways: Arc::new(gateways::Gateways::from_env()),
     });
+    gateways::spawn_refresh(Arc::clone(&state));
 
     // Bitswap ingress: sweep partial fetches a previous process left in `tmp/`,
     // then promote/fill on a timer.
@@ -149,5 +182,5 @@ async fn main() -> Result<()> {
     }
     ingress_commit::spawn(Arc::clone(&state));
 
-    serve_transport(Arc::new(plugin::IpfsPlugin { state }), cfg.http_addr).await
+    serve_transport(Arc::new(plugin::IpfsPlugin { state, config: plane }), cfg.http_addr).await
 }

src/plane.rs

@@ -0,0 +1,173 @@
+//! The plugin's own settings (config plane), edited from meta-share's dashboard
+//! (Plugins → ipfs → configure) and stored in `<state dir>/config.json`.
+//!
+//! Every knob here was — and still is — an env var read by the module that uses
+//! it (`config.rs`, `swarm/`, `ingress_commit.rs`). Rather than re-thread each
+//! one, [`overlay_env`] writes the effective values onto those same env names
+//! once at boot, before anything reads them: `config.json` wins, the env seeds
+//! it. Infra stays env-only: `PUBLIC_ADDR` / `P2P_LISTEN` (bound to the port
+//! compose publishes and the box's IP) and the data dirs.
+
+use meta_feeder_sdk::transport::config::read;
+use meta_feeder_sdk::{ConfigField, ConfigSchema};
+use serde_json::{json, Value};
+
+const GIB: u64 = 1024 * 1024 * 1024;
+const DEFAULT_NAMESPACE: &str = "metamesh-share-default";
+
+pub fn schema() -> ConfigSchema {
+    ConfigSchema {
+        fields: vec![
+            ConfigField::text("kad_namespace", "Cohort namespace")
+                .with_help("Peers only see others with the same namespace. Change it on every node of a private group."),
+            ConfigField::list("kad_bootstrap_peers", "DHT bootstrap peers")
+                .with_help("Multiaddrs used to join the public IPFS DHT. Empty = the app's default list. `none` disables the DHT."),
+            ConfigField::list("bootstrap_peers", "Pinned peers")
+                .with_help("Multiaddrs this node keeps dialled (e.g. a known gateway)."),
+            ConfigField::list("gateway_urls", "Local gateways")
+                .with_help("Base URLs of gateways reached over HTTP without the swarm (e.g. http://metagateway-app:3000). They resolve NZB and subtitle pointers even when the libp2p link is down."),
+            ConfigField::bool("enable_mdns", "LAN discovery (mDNS)"),
+            ConfigField::bool("seed_dht_provide", "Announce seeds on the DHT")
+                .with_help("Publish provider records for what this node seeds, so remote peers can find it."),
+            ConfigField::number("max_conns", "Max connections"),
+            ConfigField::bool("ingress_commit", "Background completion")
+                .with_help("Finish partially fetched IPFS content in the background."),
+            ConfigField::number("ingress_commit_max_concurrent", "Concurrent completions"),
+            ConfigField::number("ingress_commit_min_free_gib", "Disk floor (GiB)")
+                .with_help("Background completion stands down below this much free disk."),
+        ],
+    }
+}
+
+fn env(key: &str) -> Option<String> {
+    std::env::var(key).ok().map(|s| s.trim().to_string()).filter(|s| !s.is_empty())
+}
+
+fn env_list(key: &str) -> Vec<String> {
+    env(key)
+        .map(|s| s.split(',').map(|p| p.trim().to_string()).filter(|p| !p.is_empty()).collect())
+        .unwrap_or_default()
+}
+
+fn env_flag(key: &str, default: bool) -> bool {
+    match env(key) {
+        Some(v) => !matches!(v.to_ascii_lowercase().as_str(), "0" | "false" | "no" | "off"),
+        None => default,
+    }
+}
+
+fn env_u64(key: &str, default: u64) -> u64 {
+    env(key).and_then(|v| v.parse().ok()).unwrap_or(default)
+}
+
+/// The values a plugin with no `config.json` runs with. Booleans are explicit
+/// (the page's checkbox always submits one), so each carries the default the
+/// reading module would apply.
+pub fn seed() -> Value {
+    // `config.rs` defaults the seed announce to "public mode", i.e. a PUBLIC_ADDR.
+    let public = env("PUBLIC_ADDR").or_else(|| env("P2P_ANNOUNCE")).is_some();
+    json!({
+        "kad_namespace": env("KAD_NAMESPACE").unwrap_or_else(|| DEFAULT_NAMESPACE.to_string()),
+        "kad_bootstrap_peers": env_list("KAD_BOOTSTRAP_PEERS"),
+        "bootstrap_peers": env_list("BOOTSTRAP_PEERS"),
+        "gateway_urls": env_list("META_SHARE_GATEWAY_URLS"),
+        "enable_mdns": env_flag("ENABLE_MDNS", true),
+        "seed_dht_provide": env_flag("META_SHARE_SEED_DHT_PROVIDE", public),
+        "max_conns": env_u64("META_SHARE_MAX_CONNS", 128),
+        "ingress_commit": env_flag("META_SHARE_INGRESS_COMMIT", true),
+        "ingress_commit_max_concurrent": env_u64("META_SHARE_INGRESS_COMMIT_MAX_CONCURRENT", 2),
+        "ingress_commit_min_free_gib": env_u64("META_SHARE_INGRESS_COMMIT_MIN_FREE_BYTES", 5 * GIB) / GIB,
+    })
+}
+
+/// `(env name, value)` pairs the effective config sets. An unset/blank value
+/// is skipped, leaving the env (or the module's default) in force.
+pub fn env_overrides(v: &Value) -> Vec<(&'static str, String)> {
+    let mut out = Vec::new();
+    let bool_s = |b: bool| if b { "true".to_string() } else { "false".to_string() };
+    if let Some(s) = read::text(v, "kad_namespace") {
+        out.push(("KAD_NAMESPACE", s));
+    }
+    if let Some(l) = read::list(v, "kad_bootstrap_peers").filter(|l| !l.is_empty()) {
+        // `none` is the documented "no DHT" spelling; config.rs reads set-but-empty that way.
+        let joined = if l.iter().any(|p| p.eq_ignore_ascii_case("none")) { String::new() } else { l.join(",") };
+        out.push(("KAD_BOOTSTRAP_PEERS", joined));
+    }
+    if let Some(l) = read::list(v, "bootstrap_peers").filter(|l| !l.is_empty()) {
+        out.push(("BOOTSTRAP_PEERS", l.join(",")));
+    }
+    if let Some(l) = read::list(v, "gateway_urls").filter(|l| !l.is_empty()) {
+        out.push(("META_SHARE_GATEWAY_URLS", l.join(",")));
+    }
+    if let Some(b) = read::flag(v, "enable_mdns") {
+        out.push(("ENABLE_MDNS", bool_s(b)));
+    }
+    if let Some(b) = read::flag(v, "seed_dht_provide") {
+        out.push(("META_SHARE_SEED_DHT_PROVIDE", bool_s(b)));
+    }
+    if let Some(n) = read::uint(v, "max_conns").filter(|n| *n > 0) {
+        out.push(("META_SHARE_MAX_CONNS", n.to_string()));
+    }
+    if let Some(b) = read::flag(v, "ingress_commit") {
+        out.push(("META_SHARE_INGRESS_COMMIT", bool_s(b)));
+    }
+    if let Some(n) = read::uint(v, "ingress_commit_max_concurrent").filter(|n| *n > 0) {
+        out.push(("META_SHARE_INGRESS_COMMIT_MAX_CONCURRENT", n.to_string()));
+    }
+    if let Some(n) = read::uint(v, "ingress_commit_min_free_gib") {
+        out.push(("META_SHARE_INGRESS_COMMIT_MIN_FREE_BYTES", n.saturating_mul(GIB).to_string()));
+    }
+    out
+}
+
+/// Apply [`env_overrides`] to this process's environment. Call once at the top
+/// of `main`, before any config is read and before other tasks run.
+pub fn overlay_env(v: &Value) {
+    for (k, val) in env_overrides(v) {
+        std::env::set_var(k, val);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn get<'a>(o: &'a [(&'static str, String)], k: &str) -> Option<&'a str> {
+        o.iter().find(|(n, _)| *n == k).map(|(_, v)| v.as_str())
+    }
+
+    #[test]
+    fn page_values_map_onto_the_env_names_the_modules_read() {
+        let o = env_overrides(&json!({
+            "kad_namespace": "private-cohort",
+            "kad_bootstrap_peers": ["/dns4/a/tcp/4001/p2p/X", "/dns4/b/tcp/4001/p2p/Y"],
+            "bootstrap_peers": [],
+            "enable_mdns": false,
+            "seed_dht_provide": true,
+            "max_conns": null,
+            "ingress_commit": true,
+            "ingress_commit_max_concurrent": 4,
+            "ingress_commit_min_free_gib": 10,
+        }));
+        assert_eq!(get(&o, "KAD_NAMESPACE"), Some("private-cohort"));
+        assert_eq!(get(&o, "KAD_BOOTSTRAP_PEERS"), Some("/dns4/a/tcp/4001/p2p/X,/dns4/b/tcp/4001/p2p/Y"));
+        assert_eq!(get(&o, "BOOTSTRAP_PEERS"), None, "an empty list keeps the env");
+        assert_eq!(get(&o, "ENABLE_MDNS"), Some("false"));
+        assert_eq!(get(&o, "META_SHARE_MAX_CONNS"), None, "blank number keeps the env");
+        assert_eq!(get(&o, "META_SHARE_INGRESS_COMMIT_MIN_FREE_BYTES"), Some("10737418240"));
+    }
+
+    #[test]
+    fn none_disables_the_dht() {
+        let o = env_overrides(&json!({ "kad_bootstrap_peers": ["none"] }));
+        assert_eq!(get(&o, "KAD_BOOTSTRAP_PEERS"), Some(""));
+    }
+
+    #[test]
+    fn every_seeded_key_is_in_the_schema() {
+        let keys: Vec<String> = schema().fields.into_iter().map(|f| f.key).collect();
+        for k in seed().as_object().unwrap().keys() {
+            assert!(keys.contains(k), "{k} missing from schema");
+        }
+    }
+}

src/plugin.rs

@@ -17,11 +17,11 @@ use meta_feeder_sdk::transport::ipfs::{
     RelPath, ShareContainer, ShareLibrary, Shared,
 };
 use meta_feeder_sdk::transport::{
-    ApiError, Capabilities, Deleted, FocusView, Health, Job, Lane, Manifest, ReconcileReport,
+    ApiError, Capabilities, ConfigPlane, Deleted, FocusView, Health, Job, Lane, Manifest, ReconcileReport,
     ReconcileRequest, TransportPlugin, CONTRACT_VERSION,
 };
 use serde::Deserialize;
-use tracing::debug;
+use tracing::{debug, warn};
 
 use crate::api::files::{bitswap, bytes_to_response, PLAYER_HEADER};
 use crate::api::AppState;
@@ -34,6 +34,7 @@ const IMPORT_BODY_LIMIT: usize = 600 * 1024 * 1024;
 
 pub struct IpfsPlugin {
     pub state: Arc<AppState>,
+    pub config: Arc<ConfigPlane>,
 }
 
 fn lane_of(headers: &HeaderMap) -> Lane {
@@ -53,6 +54,7 @@ impl TransportPlugin for IpfsPlugin {
             version: env!("CARGO_PKG_VERSION").into(),
             contract: CONTRACT_VERSION,
             capabilities: Capabilities { fetch: true, share: true },
+            config: true,
         }
     }
 
@@ -110,9 +112,14 @@ impl TransportPlugin for IpfsPlugin {
         &self.state.focus
     }
 
+    fn config(&self) -> Option<Arc<ConfigPlane>> {
+        Some(Arc::clone(&self.config))
+    }
+
     fn extra_routes(self: Arc<Self>) -> Router {
         let gateway = Router::new()
             .route("/ipfs/:cid", get(crate::api::ipfs_gateway::get_ipfs))
+            .route("/ipfs-tier/resolve/:cid", get(crate::resolve::resolve))
             .with_state(Arc::clone(&self.state));
         Router::new()
             .route("/ipfs-tier/peers", get(peers))
@@ -144,6 +151,7 @@ impl TransportPlugin for IpfsPlugin {
             .route("/ipfs-tier/share/container", post(share_container))
             .route("/ipfs-tier/stats/blockstore", get(blockstore_stats))
             .route("/ipfs-tier/debug/:cid", get(debug_block))
+            .route("/ipfs-tier/restart", post(restart))
             .with_state(self)
             .merge(gateway)
     }
@@ -209,26 +217,28 @@ async fn cat(State(p): S, Path(cid): Path<String>, Query(q): Query<CatQuery>) ->
         Err(e) => return bad_cid(e),
     };
     let max = q.max.unwrap_or_else(crate::api::ipfs_walker::max_body_bytes);
-    let block = match crate::api::ipfs_walker::get_block(&p.state, &mscid, Lane::Focused).await {
-        Ok(b) => b,
-        Err(e) => return ApiError::Upstream(format!("{e:#}")).into_response(),
-    };
-    let bytes = if mscid.codec() == crate::api::ipfs_walker::DAGPB_CODEC {
-        match crate::api::ipfs_walker::assemble_dagpb_file(&p.state, &block, max, Lane::Focused).await {
-            Ok(b) => b,
-            Err(e) => return ApiError::Upstream(format!("{e:#}")).into_response(),
-        }
-    } else {
-        if block.len() > max {
-            return ApiError::Upstream(format!(
-                "{cid} is {} bytes, over the {max}-byte ceiling",
-                block.len()
-            ))
-            .into_response();
-        }
-        block
-    };
-    (StatusCode::OK, bytes).into_response()
+    match read_object(&p.state, &mscid, max).await {
+        Ok(bytes) => (StatusCode::OK, bytes).into_response(),
+        Err(e) => ApiError::Upstream(format!("{e:#}")).into_response(),
+    }
+}
+
+/// A block, or a dag-pb file assembled, up to `max` bytes — local, else bitswap.
+pub(crate) async fn read_object(
+    state: &AppState,
+    mscid: &crate::blockstore::MsCid,
+    max: usize,
+) -> anyhow::Result<Vec<u8>> {
+    let block = crate::api::ipfs_walker::get_block(state, mscid, Lane::Focused).await?;
+    if mscid.codec() == crate::api::ipfs_walker::DAGPB_CODEC {
+        return crate::api::ipfs_walker::assemble_dagpb_file(state, &block, max, Lane::Focused).await;
+    }
+    anyhow::ensure!(
+        block.len() <= max,
+        "{mscid} is {} bytes, over the {max}-byte ceiling",
+        block.len()
+    );
+    Ok(block)
 }
 
 #[derive(Deserialize)]
@@ -422,3 +432,15 @@ async fn debug_block(State(p): S, Path(cid): Path<String>) -> Response {
 pub fn bytes_response(bytes: Vec<u8>, range: Option<&HeaderValue>) -> Response {
     bytes_to_response(bytes, range)
 }
+
+/// The hull changed the endpoints this plugin announces (`peer_url`) or reads
+/// through (`meta_core_url`): exit so `restart: unless-stopped` re-execs us and
+/// `main` asks the hull again.
+async fn restart() -> StatusCode {
+    tokio::spawn(async {
+        tokio::time::sleep(Duration::from_millis(300)).await;
+        warn!("restarting to pick up the hull's new network settings");
+        std::process::exit(0);
+    });
+    StatusCode::NO_CONTENT
+}

src/resolve.rs

@@ -0,0 +1,353 @@
+//! `GET /ipfs-tier/resolve/:cid?hint=` — a [pointer](meta_feeder_sdk::transport::pointer)'s bytes.
+//!
+//! A pointer (an `nzb-release`'s `.nzb`, a `provider-file`'s subtitle) costs
+//! quota on an account this peer does not hold, so only a gateway whose feeder
+//! holds the key can answer it — never bitswap: the cid is an address, not a
+//! hash of the bytes.
+//!
+//! 1. **Stored.** `hint` is the content cid the record's pointer field names
+//!    (the hull reads it). Held locally → served, no network.
+//! 2. **Redeemed.** Otherwise `POST {gateway}/api/file/:cid/redeem` at each
+//!    gateway whose claim covers the pointer, fastest first
+//!    ([`crate::gateways`]). The gateway names the bytes' content cid; we
+//!    re-derive it with the kubo-identical chunker and refuse a mismatch, then
+//!    store the blocks. A repeat redeem is answered from the gateway's own store
+//!    without spending quota, so a second box pays nothing.
+//!
+//! The hull owns what follows — the seed row, the permanent copy in meta-core,
+//! the record link — so this route only fetches, verifies and stores blocks.
+//! Moved here from the hull's `redeem.rs` (meta-share 2.4.0), unchanged in
+//! behaviour: ordering, timeout, size cap, verification, error ranking.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use axum::extract::{Path, Query, State};
+use axum::http::{header::RETRY_AFTER, HeaderValue, StatusCode};
+use axum::response::{IntoResponse, Response};
+use meta_feeder_sdk::transport::dto::{
+    HDR_RESOLVE_ERROR, RESOLVE_NOT_FOUND, RESOLVE_NO_GATEWAY, RESOLVE_QUOTA, RESOLVE_UNCLAIMED,
+    RESOLVE_UPSTREAM,
+};
+use meta_feeder_sdk::transport::hull::{HDR_MANIFEST_CID, HDR_MANIFEST_SOURCE};
+use meta_feeder_sdk::transport::pointer;
+use serde::Deserialize;
+use tracing::{debug, warn};
+
+use crate::api::AppState;
+use crate::blockstore::parse_record_cid;
+use crate::ipfs_chunk::IpfsBlocks;
+
+/// Response header naming the redeemed bytes' content cid.
+pub const CONTENT_CID_HEADER: &str = "x-metamesh-content-cid";
+
+/// One redeem's wall-clock budget: the gateway makes a metered upstream call and
+/// a store-back before it answers.
+const REDEEM_TIMEOUT: Duration = Duration::from_secs(60);
+
+/// Ceiling on a redeemed body — the gateway's own `MAX_REDEEMED_BYTES`. A
+/// 100k-segment `.nzb` is ~16 MiB and a subtitle kilobytes, but a comic chapter
+/// redeems as one CBZ (`tachiyomi`, meta-feeder-suwayomi): ~24 MB measured for a
+/// 24-page chapter, more for a long webtoon one. Bounds a broken or hostile
+/// gateway. ⚠ The hull's resolve client caps the same body
+/// (`meta-share/src/plugins.rs`, `resolve`); raise both together.
+pub const MAX_REDEEM_BYTES: usize = 64 * 1024 * 1024;
+
+#[derive(Debug)]
+pub enum RedeemError {
+    /// Every claiming gateway answered `404`.
+    NotFound,
+    /// A gateway refused on quota; `retry_after` in seconds.
+    Quota { retry_after: Option<u64> },
+    /// Transport failure, unexpected status, or a body that doesn't hash to the
+    /// cid the gateway named.
+    Upstream(String),
+}
+
+#[derive(Deserialize)]
+pub struct ResolveQuery {
+    #[serde(default)]
+    hint: Option<String>,
+}
+
+pub async fn resolve(
+    State(state): State<Arc<AppState>>,
+    Path(cid): Path<String>,
+    Query(q): Query<ResolveQuery>,
+) -> Response {
+    let ptr = match pointer::decode(&cid) {
+        Some(Ok(p)) => p,
+        Some(Err(e)) => return refuse(StatusCode::BAD_REQUEST, RESOLVE_NOT_FOUND, None, format!("{e:#}")),
+        None => {
+            return refuse(
+                StatusCode::BAD_REQUEST,
+                RESOLVE_NOT_FOUND,
+                None,
+                format!("`{cid}` is not a pointer cid"),
+            )
+        }
+    };
+
+    if let Some(hint) = q.hint.as_deref().filter(|h| !h.is_empty()) {
+        if let Some(bytes) = stored(&state, hint).await {
+            debug!(cid, hint, "resolve: served the stored copy");
+            return answer(bytes, "store", hint);
+        }
+    }
+
+    let Some(key) = ptr.redeem else {
+        return refuse(
+            StatusCode::NOT_FOUND,
+            RESOLVE_NOT_FOUND,
+            None,
+            format!("`{cid}` resolves only from a stored copy, and none is held here"),
+        );
+    };
+    let targets = state.gateways.candidates(&state.peer_directory, key.codec, &key.key);
+    if targets.is_empty() {
+        return if state.gateways.knows_any(&state.peer_directory) {
+            refuse(
+                StatusCode::NOT_FOUND,
+                RESOLVE_UNCLAIMED,
+                None,
+                format!("no known gateway redeems {} `{}`", key.codec, key.key),
+            )
+        } else {
+            refuse(
+                StatusCode::SERVICE_UNAVAILABLE,
+                RESOLVE_NO_GATEWAY,
+                Some(2),
+                "no gateway reachable yet".into(),
+            )
+        };
+    }
+
+    match redeem_from(&state.http, &targets, &cid).await {
+        Ok((blocks, body)) => {
+            let root = blocks.root.clone();
+            if let Err(e) = crate::blockstore::put_ipfs_blocks(state.block_store.as_ref(), &blocks).await {
+                // The bytes are verified; failing to keep them only costs a
+                // repeat redeem, which the gateway answers from its store.
+                warn!(cid, root, error = %format!("{e:#}"), "resolve: storing the redeemed blocks failed");
+            }
+            debug!(cid, root, size = body.len(), "resolve: redeemed through a gateway");
+            answer(body, "redeem", &root)
+        }
+        Err(RedeemError::NotFound) => refuse(
+            StatusCode::NOT_FOUND,
+            RESOLVE_NOT_FOUND,
+            None,
+            "every gateway that claims it answered not found".into(),
+        ),
+        Err(RedeemError::Quota { retry_after }) => refuse(
+            StatusCode::SERVICE_UNAVAILABLE,
+            RESOLVE_QUOTA,
+            Some(retry_after.unwrap_or(3600)),
+            "the provider's quota is spent".into(),
+        ),
+        Err(RedeemError::Upstream(e)) => refuse(StatusCode::BAD_GATEWAY, RESOLVE_UPSTREAM, None, e),
+    }
+}
+
+/// The stored copy `hint` names, when its whole dag is held locally.
+async fn stored(state: &AppState, hint: &str) -> Option<Vec<u8>> {
+    let mscid = parse_record_cid(hint).ok()?;
+    if !crate::share::local_dag_complete(state, &mscid).await {
+        return None;
+    }
+    crate::plugin::read_object(state, &mscid, MAX_REDEEM_BYTES).await.ok()
+}
+
+fn answer(bytes: Vec<u8>, source: &'static str, root: &str) -> Response {
+    let mut r = (StatusCode::OK, bytes).into_response();
+    r.headers_mut().insert(HDR_MANIFEST_SOURCE, HeaderValue::from_static(source));
+    if let Ok(v) = HeaderValue::from_str(root) {
+        r.headers_mut().insert(HDR_MANIFEST_CID, v);
+    }
+    r
+}
+
+fn refuse(status: StatusCode, why: &'static str, retry_after: Option<u64>, msg: String) -> Response {
+    let mut r = (status, msg).into_response();
+    r.headers_mut().insert(HDR_RESOLVE_ERROR, HeaderValue::from_static(why));
+    if let Some(s) = retry_after {
+        r.headers_mut().insert(RETRY_AFTER, HeaderValue::from(s));
+    }
+    r
+}
+
+/// Try each target in turn and return the first verified answer.
+///
+/// A `404` moves on silently. A quota refusal or an upstream failure also moves
+/// on (another gateway may hold its own quota) but is remembered: when nobody
+/// delivers, quota outranks an upstream error, which outranks not-found.
+pub(crate) async fn redeem_from(
+    http: &reqwest::Client,
+    targets: &[String],
+    cid: &str,
+) -> Result<(IpfsBlocks, Vec<u8>), RedeemError> {
+    let mut quota: Option<Option<u64>> = None;
+    let mut upstream: Option<String> = None;
+    for base in targets {
+        match redeem_one(http, base, cid).await {
+            Ok(v) => return Ok(v),
+            Err(RedeemError::Quota { retry_after }) => {
+                quota.get_or_insert(retry_after);
+            }
+            Err(RedeemError::Upstream(e)) => {
+                debug!(cid, gateway = %base, error = %e, "redeem: gateway failed; trying the next");
+                upstream.get_or_insert(e);
+            }
+            Err(RedeemError::NotFound) => {}
+        }
+    }
+    if let Some(retry_after) = quota {
+        return Err(RedeemError::Quota { retry_after });
+    }
+    if let Some(e) = upstream {
+        return Err(RedeemError::Upstream(e));
+    }
+    Err(RedeemError::NotFound)
+}
+
+async fn redeem_one(
+    http: &reqwest::Client,
+    base: &str,
+    cid: &str,
+) -> Result<(IpfsBlocks, Vec<u8>), RedeemError> {
+    let url = format!("{}/api/file/{}/redeem", base.trim_end_matches('/'), cid);
+    let resp = http
+        .post(&url)
+        .timeout(REDEEM_TIMEOUT)
+        .send()
+        .await
+        .map_err(|e| RedeemError::Upstream(format!("POST {url}: {e}")))?;
+    match resp.status() {
+        s if s.is_success() => {}
+        reqwest::StatusCode::NOT_FOUND => return Err(RedeemError::NotFound),
+        reqwest::StatusCode::TOO_MANY_REQUESTS => {
+            let retry_after = resp
+                .headers()
+                .get(reqwest::header::RETRY_AFTER)
+                .and_then(|v| v.to_str().ok())
+                .and_then(|v| v.trim().parse().ok());
+            return Err(RedeemError::Quota { retry_after });
+        }
+        s => return Err(RedeemError::Upstream(format!("POST {url} answered {s}"))),
+    }
+    let advertised = resp
+        .headers()
+        .get(CONTENT_CID_HEADER)
+        .and_then(|v| v.to_str().ok())
+        .map(|v| v.trim().to_string())
+        .filter(|v| !v.is_empty())
+        .ok_or_else(|| RedeemError::Upstream(format!("POST {url}: no {CONTENT_CID_HEADER} header")))?;
+    if resp.content_length().is_some_and(|n| n > MAX_REDEEM_BYTES as u64) {
+        return Err(RedeemError::Upstream(format!(
+            "POST {url}: body over the {MAX_REDEEM_BYTES}-byte ceiling"
+        )));
+    }
+    let body = resp
+        .bytes()
+        .await
+        .map_err(|e| RedeemError::Upstream(format!("POST {url}: body: {e}")))?
+        .to_vec();
+    if body.len() > MAX_REDEEM_BYTES {
+        return Err(RedeemError::Upstream(format!(
+            "POST {url}: body over the {MAX_REDEEM_BYTES}-byte ceiling"
+        )));
+    }
+    // Kubo-identical chunking is CPU-bound; keep it off the async runtime.
+    let (blocks, body) =
+        tokio::task::spawn_blocking(move || (crate::ipfs_chunk::compute_ipfs_blocks(&body), body))
+            .await
+            .map_err(|e| RedeemError::Upstream(format!("ipfs chunker task: {e}")))?;
+    if blocks.root != advertised {
+        // Without this any gateway could answer with unrelated bytes and we
+        // would store them as the subtitle or the `.nzb`.
+        return Err(RedeemError::Upstream(format!(
+            "POST {url}: body hashes to {} but the gateway named {advertised}",
+            blocks.root
+        )));
+    }
+    Ok((blocks, body))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use axum::http::HeaderMap;
+    use axum::routing::post;
+
+    const BODY: &[u8] = b"1\n00:00:01,000 --> 00:00:02,000\nhi\n";
+
+    /// A loopback gateway: `/ok` answers truthfully, `/lie` names the wrong cid,
+    /// `/quota` refuses, `/missing` 404s.
+    async fn gateway() -> String {
+        async fn answer(bytes: &'static [u8], cid: String) -> (HeaderMap, Vec<u8>) {
+            let mut h = HeaderMap::new();
+            h.insert(CONTENT_CID_HEADER, HeaderValue::from_str(&cid).unwrap());
+            (h, bytes.to_vec())
+        }
+        let good = crate::ipfs_chunk::compute_ipfs_blocks(BODY).root;
+        let other = crate::ipfs_chunk::compute_ipfs_blocks(b"something else").root;
+        let app = axum::Router::new()
+            .route("/ok/api/file/:cid/redeem", post(move || answer(BODY, good.clone())))
+            .route("/lie/api/file/:cid/redeem", post(move || answer(BODY, other.clone())))
+            .route(
+                "/quota/api/file/:cid/redeem",
+                post(|| async {
+                    let mut h = HeaderMap::new();
+                    h.insert(RETRY_AFTER, HeaderValue::from_static("120"));
+                    (StatusCode::TOO_MANY_REQUESTS, h, "quota")
+                }),
+            )
+            .route("/missing/api/file/:cid/redeem", post(|| async { StatusCode::NOT_FOUND }));
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let base = format!("http://{}", listener.local_addr().unwrap());
+        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+        base
+    }
+
+    const LOCATOR: &str = "bagfcaaa2bvxxazloon2we5djorwgk43gnfwgkorxga3dcobtgq";
+
+    #[tokio::test]
+    async fn a_verified_answer_comes_back_with_its_blocks() {
+        let gw = gateway().await;
+        let (blocks, body) = redeem_from(&reqwest::Client::new(), &[format!("{gw}/ok")], LOCATOR)
+            .await
+            .expect("redeemed");
+        assert_eq!(body, BODY);
+        assert_eq!(blocks.root, crate::ipfs_chunk::compute_ipfs_blocks(&body).root);
+    }
+
+    #[tokio::test]
+    async fn a_body_that_does_not_hash_to_the_named_cid_is_refused() {
+        let gw = gateway().await;
+        match redeem_from(&reqwest::Client::new(), &[format!("{gw}/lie")], LOCATOR).await {
+            Err(RedeemError::Upstream(e)) => assert!(e.contains("hashes to"), "{e}"),
+            other => panic!("expected an upstream refusal, got {:?}", other.map(|_| ())),
+        }
+    }
+
+    #[tokio::test]
+    async fn a_not_found_moves_on_to_the_next_gateway() {
+        let gw = gateway().await;
+        let targets = [format!("{gw}/missing"), format!("{gw}/ok")];
+        assert!(redeem_from(&reqwest::Client::new(), &targets, LOCATOR).await.is_ok());
+    }
+
+    #[tokio::test]
+    async fn quota_outranks_not_found_when_nobody_delivers() {
+        let gw = gateway().await;
+        let targets = [format!("{gw}/missing"), format!("{gw}/quota")];
+        match redeem_from(&reqwest::Client::new(), &targets, LOCATOR).await {
+            Err(RedeemError::Quota { retry_after }) => assert_eq!(retry_after, Some(120)),
+            other => panic!("expected quota, got {:?}", other.map(|_| ())),
+        }
+        match redeem_from(&reqwest::Client::new(), &[format!("{gw}/missing")], LOCATOR).await {
+            Err(RedeemError::NotFound) => {}
+            other => panic!("expected not found, got {:?}", other.map(|_| ())),
+        }
+    }
+}

src/settings.rs

@@ -1,11 +1,9 @@
-//! The `network` section of the hull's `settings.json`, read-only.
+//! The hull's two endpoints this plugin needs: `peer_url` (announced in
+//! identify as `baseUrl=`) and `meta_core_url` (library byte reads).
 //!
-//! The hull owns the file (it seeds it from the deprecated env vars on first
-//! boot and edits it from the dashboard); this plugin only needs the two
-//! endpoints, and falls back to the same env-derived defaults the hull would
-//! seed when the file doesn't exist yet.
-
-use std::path::Path;
+//! `main` asks the hull (`GET /internal/network`) at boot; when it does not
+//! answer, [`NetworkSettings::from_env`] recomputes the same env-derived values
+//! the hull seeds its own settings from.
 
 use serde::Deserialize;
 use tracing::warn;
@@ -21,20 +19,6 @@ pub struct NetworkSettings {
 }
 
 impl NetworkSettings {
-    /// `<config_dir>/settings.json` → `network`, else [`Self::from_env`].
-    pub fn load(config_dir: &Path) -> Self {
-        #[derive(Deserialize, Default)]
-        #[serde(default)]
-        struct File {
-            network: Option<NetworkSettings>,
-        }
-        std::fs::read(config_dir.join("settings.json"))
-            .ok()
-            .and_then(|b| serde_json::from_slice::<File>(&b).ok())
-            .and_then(|f| f.network)
-            .unwrap_or_else(Self::from_env)
-    }
-
     /// The hull's first-boot seed, recomputed: `META_CORE_URL`, and
     /// `INTERNAL_BASE_URL` → `BASE_URL` → [`DEFAULT_PEER_URL`].
     pub fn from_env() -> Self {
@@ -63,22 +47,21 @@ fn env_nonempty(key: &str) -> Option<String> {
         .filter(|s| !s.is_empty())
 }
 
+impl From<meta_feeder_sdk::transport::NetworkInfo> for NetworkSettings {
+    fn from(n: meta_feeder_sdk::transport::NetworkInfo) -> Self {
+        Self {
+            meta_core_url: n.meta_core_url,
+            peer_url: n.peer_url,
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
     #[test]
-    fn reads_the_network_section_of_the_hulls_file() {
-        let d = std::env::temp_dir().join(format!("ipfs-settings-{}", std::process::id()));
-        std::fs::create_dir_all(&d).unwrap();
-        std::fs::write(
-            d.join("settings.json"),
-            r#"{"network":{"meta_core_url":"http://mc:9000","peer_url":"https://share.example"},"usenet":{"nntp_host":"x"}}"#,
-        )
-        .unwrap();
-        let n = NetworkSettings::load(&d);
-        assert_eq!(n.meta_core_url.as_deref(), Some("http://mc:9000"));
-        assert_eq!(n.peer_url.as_deref(), Some("https://share.example"));
-        let _ = std::fs::remove_dir_all(&d);
+    fn the_env_fallback_always_has_a_peer_url() {
+        assert!(NetworkSettings::from_env().peer_url.is_some());
     }
 }

src/swarm/event_loop.rs

@@ -9,6 +9,7 @@
 //! locals (not held on `self`) so each handler method can borrow
 //! `&mut self` cleanly without conflicting with the arm futures.
 
+use std::collections::HashMap;
 use std::time::Duration;
 
 use futures::StreamExt;
@@ -25,10 +26,46 @@ use crate::gateway_discovery::{advertises_gateways, fetch_gateway_caps};
 use super::bootstrap::{addr_targets_match, redial_bootstrap, BootstrapEntry};
 use super::bitswap_client::BitswapInflight;
 use super::identify_agent::parse_base_url;
-use super::kad_discovery::{namespace_key, peer_id_from_multiaddr, KadConfig};
+use super::kad_discovery::{is_dialable_addr, namespace_key, peer_id_from_multiaddr, KadConfig};
 use super::peer_directory::PeerDirectory;
+use super::provide_queue::{ProvideQueue, DEFAULT_PROVIDE_CONCURRENCY};
 use super::{Behaviour, BehaviourEvent, Command, GatewayCapsConfig, PeersInfo};
 
+/// How often [`SwarmTask::on_gateway_keepalive_tick`] re-dials dropped
+/// gateways and re-stamps live ones. Half meta-search's 60 s: an NZB play is
+/// blocked for as long as the gateway is missing (no one else can redeem).
+const GATEWAY_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30);
+
+/// A gateway we've identified and now pin for keep-alive: its API base URL (for
+/// the periodic caps refresh that keeps it routable) plus dialable addresses
+/// (for re-pinning the connection if it drops). Persists across TTL expiry and
+/// disconnect — unlike the read-side directory entry, which `forget_peer` drops
+/// on the last close — so a gateway is never permanently forgotten once seen.
+///
+/// Ported from meta-search. Without it a gateway that restarts under the same
+/// peer id is lost for good: its mDNS record never expires, so `Discovered`
+/// never fires again, and the gateway itself never dials consumers
+/// (watch.nsl.sh 2026-10-04: every NZB answered "no gateway can redeem it"
+/// until this container was restarted).
+#[derive(Clone, Debug, PartialEq)]
+struct GatewayPin {
+    base_url: Option<String>,
+    addrs: Vec<Multiaddr>,
+}
+
+/// The pin an identify from a gateway yields, or `None` for any other peer.
+/// Only dialable addresses are kept: a pin is retried for the life of the
+/// process, so a loopback / unspecified listener would be a permanent cost.
+fn gateway_pin(agent_version: &str, listen_addrs: &[Multiaddr], base_url: Option<String>) -> Option<GatewayPin> {
+    if !advertises_gateways(agent_version) {
+        return None;
+    }
+    Some(GatewayPin {
+        base_url,
+        addrs: listen_addrs.iter().filter(|a| is_dialable_addr(a)).cloned().collect(),
+    })
+}
+
 /// All the state that lives across iterations of the event loop. `swarm`
 /// and the channel receivers stay as locals in [`SwarmTask::run`] so the
 /// `tokio::select!` arms can borrow them without conflicting with
@@ -63,6 +100,11 @@ pub(super) struct SwarmTask {
     http: reqwest::Client,
     /// Timeout + debounce for that fetch.
     gateway_caps: GatewayCapsConfig,
+    /// Gateways kept connected + routable by the keep-alive tick. See
+    /// [`GatewayPin`].
+    pinned_gateways: HashMap<PeerId, GatewayPin>,
+    /// Per-CID announcements waiting for a walk slot. See [`ProvideQueue`].
+    provides: ProvideQueue,
 }
 
 impl SwarmTask {
@@ -101,6 +143,13 @@ impl SwarmTask {
             bitswap_inflight: BitswapInflight::new(),
             http,
             gateway_caps,
+            pinned_gateways: HashMap::new(),
+            provides: ProvideQueue::new(
+                std::env::var("META_SHARE_PROVIDE_CONCURRENCY")
+                    .ok()
+                    .and_then(|v| v.trim().parse().ok())
+                    .unwrap_or(DEFAULT_PROVIDE_CONCURRENCY),
+            ),
         }
     }
 
@@ -176,6 +225,11 @@ impl SwarmTask {
         let mut kad_discovery_tick = tokio::time::interval(self.kad_discovery_interval);
         kad_discovery_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
 
+        // Keep known gateways connected + routable (re-dial dropped ones,
+        // re-fetch caps on live ones). See [`GatewayPin`].
+        let mut gateway_keepalive_tick = tokio::time::interval(GATEWAY_KEEPALIVE_INTERVAL);
+        gateway_keepalive_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+
         loop {
             tokio::select! {
                 cmd = rx.recv() => match cmd {
@@ -206,6 +260,9 @@ impl SwarmTask {
                 _ = kad_discovery_tick.tick() => {
                     self.on_kad_discovery_tick(&mut swarm, &mut kad_discovery_tick);
                 },
+                _ = gateway_keepalive_tick.tick() => {
+                    self.on_gateway_keepalive_tick(&mut swarm);
+                },
                 event = swarm.select_next_some() => {
                     self.on_swarm_event(&mut swarm, event, &mut kad_discovery_tick).await;
                 }
@@ -243,23 +300,43 @@ impl SwarmTask {
                 // Announce on the public IPFS DHT, keyed by the CID's
                 // multihash bytes — that's the kubo-compatible provider key,
                 // so a vanilla IPFS node `get_providers`ing the same content
-                // finds us. The StartProviding result is logged in
-                // `default_log`. Works in `Mode::Client` (same call the
-                // gateway makes).
+                // finds us. Queued, not fired: each announce is a DHT walk, and
+                // a burst of seeds must not become a burst of walks (see
+                // `provide_queue`). The StartProviding result is logged in
+                // `default_log`. Works in `Mode::Client`.
                 let key = RecordKey::new(&cid.hash().to_bytes());
-                match swarm.behaviour_mut().kad.start_providing(key) {
-                    Ok(qid) => debug!(?qid, %cid, "kad: providing seeded cid"),
-                    Err(e) => warn!(error = %e, %cid, "kad: start_providing(cid) failed"),
+                if self.provides.enqueue(key) {
+                    debug!(%cid, pending = self.provides.pending_len(),
+                        "kad: queued seeded cid for announce");
                 }
+                self.pump_provides(swarm);
             }
             Command::StopProviding { cid } => {
                 let key = RecordKey::new(&cid.hash().to_bytes());
+                self.provides.cancel(&key);
                 swarm.behaviour_mut().kad.stop_providing(&key);
                 debug!(%cid, "kad: stopped providing seeded cid");
             }
         }
     }
 
+    /// Start queued announcements while walk slots are free.
+    fn pump_provides(&mut self, swarm: &mut Swarm<Behaviour>) {
+        while let Some(key) = self.provides.next_to_start() {
+            match swarm.behaviour_mut().kad.start_providing(key) {
+                Ok(qid) => {
+                    self.provides.started(qid);
+                    debug!(?qid, in_flight = self.provides.in_flight_len(),
+                        pending = self.provides.pending_len(), "kad: announcing seeded cid");
+                }
+                Err(e) => {
+                    self.provides.failed_to_start();
+                    warn!(error = %e, "kad: start_providing(cid) failed");
+                }
+            }
+        }
+    }
+
     /// Fired every `kad_discovery_interval` (steady-state) or
     /// `kad_fast_discovery_interval` (below floor). Issues a fresh
     /// `get_providers` query and adapts the next tick's cadence based on
@@ -292,6 +369,71 @@ impl SwarmTask {
 
     /// Top-level swarm-event dispatch. Each branch inspects `&event` and
     /// falls through to `default_log` at the end.
+    /// Keep pinned gateways connected + routable (ported from meta-search):
+    ///
+    /// 1. **Connection drop** — re-dial a pinned gateway that isn't connected,
+    ///    directly (not through the mDNS / kad-provider dial-gates), re-seeding
+    ///    its addresses into kad first in case they were evicted. The redial
+    ///    triggers identify, which restores its base URL and redeem claims.
+    /// 2. **Directory staleness** — identify goes quiet on an idle connection,
+    ///    so the 600 s gateway TTL can lapse on a live peer. Re-fetch caps
+    ///    over HTTP to re-stamp it.
+    fn on_gateway_keepalive_tick(&mut self, swarm: &mut Swarm<Behaviour>) {
+        // Small, stable set — clone to sidestep the self/swarm borrow overlap.
+        for (pid, pin) in self.pinned_gateways.clone() {
+            if swarm.is_connected(&pid) {
+                if self
+                    .peer_directory
+                    .needs_cap_refresh(&pid, self.gateway_caps.refresh_after)
+                {
+                    if let Some(base_url) = pin.base_url {
+                        self.spawn_caps_fetch(pid, base_url);
+                    }
+                }
+            } else if pin.addrs.is_empty() {
+                // Every address it announced was undialable — most likely a
+                // gateway without PUBLIC_ADDR seen only over a relay. A dial
+                // would fail every tick for the life of the process.
+                trace!(peer_id = %pid,
+                    "gateway keep-alive: no dialable address for pinned gateway; skipping re-dial");
+            } else {
+                for addr in &pin.addrs {
+                    swarm.behaviour_mut().kad.add_address(&pid, addr.clone());
+                }
+                match swarm.dial(pid) {
+                    Ok(()) => debug!(peer_id = %pid,
+                        "gateway keep-alive: re-dialing dropped gateway"),
+                    Err(e) => debug!(peer_id = %pid, error = %e,
+                        "gateway keep-alive: re-dial failed"),
+                }
+            }
+        }
+    }
+
+    /// Fetch a gateway's capabilities over HTTP off its `base_url` and record
+    /// them in the directory. Fire-and-forget: a slow/dead gateway must not
+    /// stall the swarm loop; a failure keeps whatever was held (it ages out on
+    /// its own TTL) and is retried on the next identify / keep-alive tick.
+    fn spawn_caps_fetch(&self, pid: PeerId, base_url: String) {
+        let http = self.http.clone();
+        let dir = self.peer_directory.clone();
+        let timeout = self.gateway_caps.fetch_timeout;
+        tokio::spawn(async move {
+            match fetch_gateway_caps(&http, &base_url, timeout).await {
+                Ok(caps) => {
+                    let redeems = caps.redeem_claims();
+                    debug!(peer_id = %pid, base_url = %base_url,
+                        nzb_fetch = caps.nzb_fetch,
+                        redeem_claims = redeems.len(),
+                        "gateway caps: fetched");
+                    dir.record_gateway_caps(pid, caps.nzb_fetch, redeems);
+                }
+                Err(e) => debug!(peer_id = %pid, base_url = %base_url,
+                    error = %e, "gateway caps: fetch failed"),
+            }
+        });
+    }
+
     async fn on_swarm_event(
         &mut self,
         swarm: &mut Swarm<Behaviour>,
@@ -371,9 +513,11 @@ impl SwarmTask {
             identify::Event::Received { peer_id, info, .. }
         )) = &event {
             let pid = *peer_id;
-            let addrs: Vec<Multiaddr> = info.listen_addrs.clone();
-            for addr in addrs {
-                swarm.behaviour_mut().kad.add_address(&pid, addr);
+            // Only what a remote peer could actually dial: a container behind
+            // docker's bridge reports 0.0.0.0 / 127.0.0.1 listeners too, and
+            // the keep-alive below would retry a pinned bad address forever.
+            for addr in info.listen_addrs.iter().filter(|a| is_dialable_addr(a)) {
+                swarm.behaviour_mut().kad.add_address(&pid, addr.clone());
             }
             let base_url = parse_base_url(&info.agent_version).map(str::to_string);
             let update = self.peer_directory.upsert_from_identify(pid, base_url);
@@ -381,6 +525,15 @@ impl SwarmTask {
                 debug!(peer_id = %pid, base_url = %url,
                     "identify: learned/refreshed peer base URL");
             }
+            // Pin every identified gateway for the keep-alive tick, refreshing
+            // its addresses each time, so a dropped gateway gets re-dialed.
+            if let Some(pin) = gateway_pin(
+                &info.agent_version,
+                &info.listen_addrs,
+                self.peer_directory.base_url(&pid),
+            ) {
+                self.pinned_gateways.insert(pid, pin);
+            }
             // Gateway discovery. A non-empty `gateways=` token is emitted by
             // meta-gateway and nobody else, so it's the "this peer is a
             // gateway" signal. Its *capabilities* aren't on the wire though —
@@ -389,36 +542,16 @@ impl SwarmTask {
             // plain HTTP off the just-learned baseUrl instead (the reliable
             // source that replaced the flaky gossipsub heartbeat).
             //
-            // Fire-and-forget: a slow/dead gateway must not stall the swarm
-            // loop. Debounced by the directory so the identify burst on a
-            // multi-connection peer doesn't fan out into redundant fetches;
-            // identify's ~5min re-announce then keeps the capability fresh
-            // inside its 600s TTL, which is why there's no keep-alive tick.
+            // Debounced by the directory so the identify burst on a
+            // multi-connection peer doesn't fan out into redundant fetches.
+            // The keep-alive tick re-stamps it when identify goes quiet.
             if advertises_gateways(&info.agent_version)
                 && self
                     .peer_directory
                     .needs_cap_refresh(&pid, self.gateway_caps.refresh_after)
             {
                 if let Some(base_url) = self.peer_directory.base_url(&pid) {
-                    let http = self.http.clone();
-                    let dir = self.peer_directory.clone();
-                    let timeout = self.gateway_caps.fetch_timeout;
-                    tokio::spawn(async move {
-                        match fetch_gateway_caps(&http, &base_url, timeout).await {
-                            Ok(caps) => {
-                                let redeems = caps.redeem_claims();
-                                debug!(peer_id = %pid, base_url = %base_url,
-                                    nzb_fetch = caps.nzb_fetch,
-                                    redeem_claims = redeems.len(),
-                                    "gateway caps: fetched");
-                                dir.record_gateway_caps(pid, caps.nzb_fetch, redeems);
-                            }
-                            // Best-effort: keep whatever we held (it ages out on
-                            // its own TTL) and retry on the next identify.
-                            Err(e) => debug!(peer_id = %pid, base_url = %base_url,
-                                error = %e, "gateway caps: fetch failed"),
-                        }
-                    });
+                    self.spawn_caps_fetch(pid, base_url);
                 }
             }
         }
@@ -463,6 +596,18 @@ impl SwarmTask {
                 trace!("bitswap: inflight query resolved");
             }
         }
+        // 2d) A queued announce finished (ok or timed out): free its slot and
+        //     start the next. Only on the query's last step — until then the
+        //     walk is still dialing.
+        if let SwarmEvent::Behaviour(BehaviourEvent::Kad(
+            kad::Event::OutboundQueryProgressed {
+                id, result: QueryResult::StartProviding(_), step, ..
+            }
+        )) = &event {
+            if step.last && self.provides.finished(*id) {
+                self.pump_provides(swarm);
+            }
+        }
         // 3) kad providers — dial each new peer-id (subject to soft ceiling).
         if let SwarmEvent::Behaviour(BehaviourEvent::Kad(
             kad::Event::OutboundQueryProgressed {
@@ -556,3 +701,37 @@ fn default_log(event: SwarmEvent<BehaviourEvent>) {
         _ => {}
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn ma(s: &str) -> Multiaddr {
+        s.parse().unwrap()
+    }
+
+    #[test]
+    fn only_a_gateway_is_pinned() {
+        let addrs = [ma("/ip4/172.18.0.16/tcp/4002")];
+        assert_eq!(gateway_pin("meta-share/2.3.0 baseUrl=http://x:3000", &addrs, None), None);
+        assert!(gateway_pin("meta-gateway/1.0.41 gateways=usenet,tmdb", &addrs, None).is_some());
+    }
+
+    #[test]
+    fn a_pin_keeps_only_dialable_addresses() {
+        let addrs = [
+            ma("/ip4/127.0.0.1/tcp/4002"),
+            ma("/ip4/0.0.0.0/tcp/4002"),
+            ma("/ip4/172.18.0.16/tcp/4002"),
+            ma("/ip4/85.17.246.67/tcp/4002"),
+        ];
+        let pin = gateway_pin(
+            "meta-gateway/1.0.41 gateways=usenet",
+            &addrs,
+            Some("https://metagateway-watch.nsl.sh".into()),
+        )
+        .unwrap();
+        assert_eq!(pin.addrs, vec![addrs[2].clone(), addrs[3].clone()]);
+        assert_eq!(pin.base_url.as_deref(), Some("https://metagateway-watch.nsl.sh"));
+    }
+}

src/swarm/mod.rs

@@ -128,6 +128,7 @@ mod event_loop;
 mod identify_agent;
 mod kad_discovery;
 mod peer_directory;
+mod provide_queue;
 
 pub use bitswap_client::bitswap_get_block;
 pub use identify_agent::build_agent_version;

src/swarm/provide_queue.rs

@@ -0,0 +1,167 @@
+//! Bounded admission for per-CID DHT announcements.
+//!
+//! Every `Command::Provide` is a `kad.start_providing`, and every one of those is
+//! a full DHT walk: find the ~20 closest peers to the key, then dial each to store
+//! the record. Firing them unbounded turns a batch of seeds into a dial storm. On
+//! watch.nsl.sh (2026-10-06 18:14:46) a lapsed playback-focus lease released 45
+//! queued poster seeds in one second; 45 concurrent walks took this plugin from
+//! ~300 to 8,066 log lines a minute (6.8k failed dials) while a viewer was
+//! mid-episode.
+//!
+//! So announcements queue here and at most `max_in_flight` walks run at once.
+//! The queue is pure bookkeeping (no swarm, no I/O) so its rules are unit-tested;
+//! [`super::event_loop`] feeds it commands and query results and starts whatever
+//! [`ProvideQueue::next_to_start`] hands back.
+//!
+//! What it does NOT bound: libp2p-kad's own periodic republication of records
+//! already provided (`provider_publication_interval`), which kad paces itself.
+
+use std::collections::{HashSet, VecDeque};
+
+use libp2p::kad::{QueryId, RecordKey};
+
+/// Default number of concurrent announce walks (`META_SHARE_PROVIDE_CONCURRENCY`).
+pub(crate) const DEFAULT_PROVIDE_CONCURRENCY: usize = 2;
+
+pub(crate) struct ProvideQueue {
+    pending: VecDeque<RecordKey>,
+    queued: HashSet<RecordKey>,
+    in_flight: HashSet<QueryId>,
+    max_in_flight: usize,
+}
+
+impl ProvideQueue {
+    /// `max_in_flight == 0` is treated as 1: a queue that never drains would
+    /// silently stop announcing anything.
+    pub(crate) fn new(max_in_flight: usize) -> Self {
+        Self {
+            pending: VecDeque::new(),
+            queued: HashSet::new(),
+            in_flight: HashSet::new(),
+            max_in_flight: max_in_flight.max(1),
+        }
+    }
+
+    /// Queue `key`. `false` if it is already waiting (a re-seed of the same cid
+    /// before its turn costs nothing extra).
+    pub(crate) fn enqueue(&mut self, key: RecordKey) -> bool {
+        if !self.queued.insert(key.clone()) {
+            return false;
+        }
+        self.pending.push_back(key);
+        true
+    }
+
+    /// Drop a still-waiting `key` (its seed was removed before its turn). A walk
+    /// already in flight is left alone; the caller's `stop_providing` covers it.
+    pub(crate) fn cancel(&mut self, key: &RecordKey) {
+        if self.queued.remove(key) {
+            self.pending.retain(|k| k != key);
+        }
+    }
+
+    /// The next key to announce, if a slot is free. The caller must report the
+    /// started query with [`started`](Self::started), or [`failed_to_start`](Self::failed_to_start).
+    pub(crate) fn next_to_start(&mut self) -> Option<RecordKey> {
+        if self.in_flight.len() >= self.max_in_flight {
+            return None;
+        }
+        let key = self.pending.pop_front()?;
+        self.queued.remove(&key);
+        Some(key)
+    }
+
+    pub(crate) fn started(&mut self, qid: QueryId) {
+        self.in_flight.insert(qid);
+    }
+
+    /// `start_providing` refused synchronously (e.g. the record store is full):
+    /// nothing is in flight, so there is nothing to record. Exists so the call
+    /// site reads symmetrically.
+    pub(crate) fn failed_to_start(&mut self) {}
+
+    /// A `StartProviding` query finished. `true` if it was one of ours (the
+    /// namespace re-publish uses the same query type and is not queued).
+    pub(crate) fn finished(&mut self, qid: QueryId) -> bool {
+        self.in_flight.remove(&qid)
+    }
+
+    pub(crate) fn pending_len(&self) -> usize {
+        self.pending.len()
+    }
+
+    pub(crate) fn in_flight_len(&self) -> usize {
+        self.in_flight.len()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use libp2p::kad::{store::MemoryStore, Behaviour};
+    use libp2p::{identity, PeerId};
+
+    fn key(n: u8) -> RecordKey {
+        RecordKey::new(&[n])
+    }
+
+    /// Real `QueryId`s can only be minted by a kad behaviour; start throwaway
+    /// queries on one (nothing is ever polled, so no I/O happens).
+    fn qids(n: usize) -> Vec<QueryId> {
+        let peer = PeerId::from(identity::Keypair::generate_ed25519().public());
+        let mut kad = Behaviour::new(peer, MemoryStore::new(peer));
+        (0..n).map(|i| kad.get_providers(key(i as u8))).collect()
+    }
+
+    #[test]
+    fn at_most_max_in_flight_walks_run_at_once() {
+        let mut q = ProvideQueue::new(2);
+        for n in 0..45 {
+            assert!(q.enqueue(key(n)));
+        }
+        let ids = qids(3);
+        assert_eq!(q.next_to_start(), Some(key(0)));
+        q.started(ids[0]);
+        assert_eq!(q.next_to_start(), Some(key(1)));
+        q.started(ids[1]);
+        assert_eq!(q.next_to_start(), None, "both slots busy");
+        assert_eq!(q.pending_len(), 43);
+
+        assert!(q.finished(ids[0]));
+        assert_eq!(q.next_to_start(), Some(key(2)), "a finished walk frees its slot, FIFO");
+        q.started(ids[2]);
+        assert_eq!(q.in_flight_len(), 2);
+    }
+
+    #[test]
+    fn a_duplicate_waits_once_and_a_cancelled_key_never_starts() {
+        let mut q = ProvideQueue::new(1);
+        assert!(q.enqueue(key(1)));
+        assert!(!q.enqueue(key(1)), "already queued");
+        assert!(q.enqueue(key(2)));
+        q.cancel(&key(1));
+        assert_eq!(q.next_to_start(), Some(key(2)));
+        assert_eq!(q.pending_len(), 0);
+        // Once started (dequeued), the same cid may be queued again.
+        assert!(q.enqueue(key(2)));
+    }
+
+    #[test]
+    fn a_foreign_query_does_not_free_a_slot() {
+        let mut q = ProvideQueue::new(1);
+        q.enqueue(key(1));
+        q.enqueue(key(2));
+        let ids = qids(2);
+        q.next_to_start();
+        q.started(ids[0]);
+        assert!(!q.finished(ids[1]), "the namespace re-publish is not ours");
+        assert_eq!(q.next_to_start(), None);
+    }
+
+    #[test]
+    fn zero_concurrency_still_drains() {
+        let mut q = ProvideQueue::new(0);
+        q.enqueue(key(1));
+        assert_eq!(q.next_to_start(), Some(key(1)));
+    }
+}