take meta-feeder-sdk from gitea (github repo is gone; v1.3.3 never built)

.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 }}

Cargo.lock

@@ -1964,8 +1964,8 @@ checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98"
 
 [[package]]
 name = "meta-feeder-sdk"
-version = "1.7.0"
-source = "git+https://github.com/worph/meta-feeder-sdk?tag=v1.7.0#027917c8cffe7ca43dadcf3b426cc5f69ecc443a"
+version = "1.8.0"
+source = "git+https://gitea-watch.nsl.sh/metastar/meta-feeder-sdk?tag=v1.8.0#612f2aed661925cd22d51b29564832991e246e4d"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -1990,7 +1990,7 @@ dependencies = [
 
 [[package]]
 name = "meta-transport-ipfs"
-version = "1.3.0"
+version = "1.3.1"
 dependencies = [
  "anyhow",
  "async-trait",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "meta-transport-ipfs"
-version = "1.3.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.8.0", features = ["transport"] }
+meta-feeder-sdk = { git = "https://gitea-watch.nsl.sh/metastar/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.8.0", features = ["testkit"] }
+meta-feeder-sdk = { git = "https://gitea-watch.nsl.sh/metastar/meta-feeder-sdk", tag = "v1.8.0", features = ["testkit"] }
 
 # ---------------------------------------------------------------------------
 # Workaround for the yanked transitive `core2` (all versions). `multihash-

src/resolve.rs

@@ -45,9 +45,13 @@ pub const CONTENT_CID_HEADER: &str = "x-metamesh-content-cid";
 /// a store-back before it answers.
 const REDEEM_TIMEOUT: Duration = Duration::from_secs(60);
 
-/// Ceiling on a redeemed body. A 100k-segment `.nzb` is ~16 MiB; a subtitle is
-/// kilobytes. Bounds a broken or hostile gateway.
-pub const MAX_REDEEM_BYTES: usize = 32 * 1024 * 1024;
+/// 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 {

src/swarm/event_loop.rs

@@ -28,6 +28,7 @@ use super::bitswap_client::BitswapInflight;
 use super::identify_agent::parse_base_url;
 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
@@ -102,6 +103,8 @@ pub(super) struct SwarmTask {
     /// 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 {
@@ -141,6 +144,12 @@ impl SwarmTask {
             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),
+            ),
         }
     }
 
@@ -291,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
@@ -567,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 {

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)));
+    }
+}