feat(digsex): wire the dig-sex economic layer into the node

Cargo.toml

@@ -32,7 +32,7 @@ edition = "2021"
 # the ROOT manifest (`[workspace.package].version`), so it MUST be set here for a
 # release to fire (§3.6). The library crates (dig-node-core/dig-runtime/dig-wallet)
 # keep their own independent versions — only the released binary tracks the workspace version.
-version = "0.246.0"
+version = "0.247.0"
 
 # Release hardening, matching digstore: keep integer-overflow checks ON in release.
 # The node parses untrusted serialized input and does offset/length arithmetic over

SPEC.md

@@ -3229,7 +3229,52 @@ batch's not-held → holder-hint enrichment (`Node::availability_answer`) runs t
 `find_providers` lookup per not-held item. Both spend this node's network bandwidth, so a caller who
 cannot name any content it actually wants could otherwise amplify this node by naming arbitrary
 `(store_id, root, retrieval_key)` triples — and a `getAvailability` batch is the LARGEST such vector,
-naming up to `MAX_AVAILABILITY_ITEMS` (= 512) content ids in one request. Three bounds govern the path:
+naming up to `MAX_AVAILABILITY_ITEMS` (= 512) content ids in one request. FOUR bounds govern the
+path, and the first of them (10.4.0) runs in FRONT of the per-requestor budget of 10.4.1:
+
+10.4.0. **Inbound admission on the mTLS peer surface (dig-sex SPEC 8.5, dig-node#269).** Every
+inbound `dig.getAvailability` and peer JSON-RPC request MUST pass a concurrency meter BEFORE the
+request is read, decoded or dispatched — ahead of the per-requestor token bucket of 10.4.1. A refused
+request is answered `-32000` with `message: "request refused"` and `data.reason` naming the LIMIT that
+was reached, never the standing of the peer: `unauthenticated`, `request too large`,
+`node at capacity`, `peer at capacity`, `relay budget exhausted`, `meter full`. A second
+implementation MUST produce these answers and MUST be able to interpret them; they say "retry later"
+(or, for the first two, "this request is not admissible as framed"), never "you are banned".
+
+- **Metered by the authenticated identity.** The meter key MUST be the mTLS-verified `peer_id` of the
+  session, as lowercase 64-hex. A session carrying no such identity is REFUSED (`unauthenticated`) —
+  never admitted unmetered, and never coerced into a placeholder key. Admitting an identity-less
+  request unmetered would make presenting no identity the cheapest way out of the meter, and metering
+  every such request under one shared key would let a single caller exhaust the allowance of everyone.
+  A caller-less session therefore serves only the range and module-range paths.
+- **A batch past `MAX_AVAILABILITY_ITEMS` (= 512) is refused WHOLE**, with reason `request too large`,
+  rather than answered as a truncated 512-item prefix. The clamp is on the quantity the caller chose,
+  applied at the boundary, and it MUST equal the batch size the node advertises it answers: a clamp set
+  below the advertised limit would refuse work this contract says is served. A batch AT 512 MUST be
+  answered in full.
+- **Two pools, and the property they buy.** The FIRST concurrent unit of work of a peer is charged to a
+  reserve whose per-peer share is exactly 1, sized `RESERVED_FIRST_SLOTS` = `MAX_INFLIGHT_PEER_CONNECTIONS`
+  (= 512); every further concurrent unit of that peer, and all relayed work, draws on the shared
+  node-wide pool. The reserve grants no peer any extra concurrency — the total concurrent share of a
+  peer is unchanged, and only the pool its first unit is charged to differs.
+
+  The normative property is this: **a bounded number of free identities MUST NOT be able to deny the
+  peer surface to everyone else.** With a single shared pool, the identities needed to hold the
+  node-wide ceiling is `global_ceiling / per_peer_share` — a small constant, each identity costing one
+  self-signed keypair and each staying inside its own share so the per-peer limiter never fires.
+  Reserving the first unit makes the cost of denying an honest peer **one identity AND one held
+  connection per slot** — linear, and bounded by the connection cap the node already enforces. An
+  implementation MAY choose different numbers; it MUST NOT make denial cheaper than one held connection
+  per denied slot.
+- **A peer holding no work in flight is admitted while a busy node sheds**, for up to
+  `RESERVED_FIRST_SLOTS` such peers concurrently. Shedding under load MUST
+  come out of the shared pool, so load-shedding degrades the peers that are already consuming
+  concurrency rather than locking out peers that are asking for the first time.
+- **The relay budget is configured and VACUOUS on this node.** Nothing here constructs relayed work, so
+  the separate relay ceiling is satisfied because the case it governs never occurs — not because it is
+  enforced. It is retained so that the first producer of relayed work inherits a budget rather than an
+  omission. It is recorded as vacuous rather than listed as an active rule, because a limit nobody
+  reaches and a limit nobody applies are indistinguishable from the number alone.
 
 10.4.1. **Per-requestor rate limit.** A token-bucket limiter (default burst
 `DEFAULT_MISS_LOOKUP_BURST` = 16, refill `DEFAULT_MISS_LOOKUP_REFILL_PER_SEC` = 4/s) sits in FRONT of

crates/dig-node-core/Cargo.toml

@@ -30,7 +30,7 @@ name = "dig-node-core"
 # dig-node#276/#296). Changing a public return type is BREAKING for an out-of-workspace implementor;
 # this crate is consumed in-workspace only and is pre-1.0, so it is a MINOR bump under SemVer's 0.x
 # rule -- recorded here rather than letting the number imply the locator surface held still.
-version = "0.66.0"
+version = "0.67.0"
 edition = "2021"
 license = "GPL-2.0-only"
 description = "The canonical DIG node ENGINE library (crate `dig_node_core`): the JSON-RPC dispatch (`handle_rpc`, the same contract as rpc.dig.net), local-first content serve/fetch/redirect from LOCAL .dig store modules (via digstore_host::serve_blind), chain-anchored-root resolution, chain-watch + subscriptions + generation gap-fill, the LRU cache, and the full P2P stack. Shared UNCHANGED by both host shells: the `dig-node` OS-service binary (dig-node-service) and the DIG Browser's in-process cdylib (dig-runtime). Native Rust so the compiled-module serve path works."

crates/dig-node-core/src/download.rs

@@ -898,6 +898,15 @@ pub struct NodeContent {
     /// at all, because pool membership is its liveness gate. Folding them together would make one
     /// structure whose key, lifetime and eviction rule all mean two different things at once.
     ask_routing: AskRoutingState,
+    /// What this node has observed each pool peer DO, and the dial share that earns them (#268).
+    ///
+    /// Held beside [`Self::ask_routing`] because the two read the same verified identity and are
+    /// bounded by the same pool: routing ranks the peers worth asking FIRST, conduct decides which are
+    /// worth asking AT ALL. Node-local and never gossiped — a shared reputation channel would be a
+    /// defamation primitive.
+    conduct: crate::seams::dig_peer::conduct::ConductState,
+    /// The instant conduct ticks are measured from. Wall-clock-free and monotonic.
+    conduct_epoch: std::time::Instant,
 }
 
 /// What a holder search ESTABLISHED — the records it found AND whether an empty result is a fact.
@@ -1407,6 +1416,8 @@ impl NodeContent {
             ask_seen: AskSeenSet::new(),
             onion_relay: std::sync::atomic::AtomicBool::new(onion_relay_from_env()),
             ask_routing,
+            conduct: crate::seams::dig_peer::conduct::ConductState::new(),
+            conduct_epoch: std::time::Instant::now(),
         })
     }
 
@@ -1906,15 +1917,22 @@ impl NodeContent {
             .filter_map(|(peer, _)| RoutedPeer::from_pool_key(peer))
             .collect();
 
+        // CONDUCT gates who is dialable at all (#268, SPEC 8.3); routing then ranks what remains.
+        // The order matters: ranking a peer this node has PROVEN dishonest would still spend a dial on
+        // it whenever the ranking happened to favour it. A peer excluded here has a verifiable fault
+        // against it — a lie or a self-contradiction — and never merely a slow or silent history,
+        // which `dial_share` floors above zero precisely so distress cannot evict an honest holder.
+        let dialable = self.conduct.dialable(&routable, self.conduct_ticks());
+
         // Ranked by what THIS node has observed, not by the pool `HashMap`'s arbitrary order — and the
-        // observations of peers no longer in `routable` are dropped in the same call, so a cycled-away
+        // observations of peers no longer in `dialable` are dropped in the same call, so a cycled-away
         // peer leaves this node's memory when it leaves the pool.
         let decision = self.ask_routing.decide(
             &config,
             asker,
             budget.remaining(config.hop_cap),
             me,
-            &routable,
+            &dialable,
             self.relay_rate_limiter.check(requestor),
         );
 
@@ -2002,6 +2020,30 @@ impl NodeContent {
             // The ONLY writer of this node's routing memory, fed an outcome this node classified from
             // an exchange it issued and saw complete (dig_ecosystem#3129).
             self.ask_routing.record(routed, &outcome, started.elapsed());
+            // The SAME exchange, classified for conduct (#268). An outcome where the peer answered —
+            // including an honest "I do not have it" — is an `HonestAnswer`, because SPEC 8.2A
+            // requires that answering is never worse than staying silent. A refusal, a timeout and an
+            // unreachable peer are `NonPerformance`: unverifiable, decaying, and floored, since none
+            // of them can be distinguished from distress an attacker induced in an honest peer.
+            //
+            // Neither VERIFIABLE class is produced here, and deliberately so. A `ProvenLie` needs
+            // bytes that failed verification against the anchor, attributed to the peer that supplied
+            // them; that attribution happens inside `dig-download`'s engine against `chunk_hashes` and
+            // is not surfaced per-peer to this node (see the report on #268). Claiming one from a
+            // transport error would brand an honest peer on unverifiable evidence, which is the exact
+            // conflation SPEC 8.2A exists to prevent.
+            self.conduct.observe(
+                routed,
+                match outcome {
+                    AskOutcome::Answered(_) | AskOutcome::AnsweredInconclusive(_) => {
+                        dig_sex::ConductEvidence::HonestAnswer
+                    }
+                    AskOutcome::Refused | AskOutcome::TimedOut | AskOutcome::Unreachable => {
+                        dig_sex::ConductEvidence::NonPerformance
+                    }
+                },
+                self.conduct_ticks(),
+            );
             match outcome {
                 AskOutcome::Answered(records) => answers.records.extend(records),
                 // The peer answered and told us its OWN subtree did not finish. Its records are
@@ -2038,6 +2080,16 @@ impl NodeContent {
             .collect()
     }
 
+    /// This node's monotonic conduct clock, in seconds since process start.
+    ///
+    /// Conduct decay is measured in elapsed ticks and nothing else, so the clock must advance on its
+    /// own — a counter incremented per exchange would mean a peer nobody dials never ages out of its
+    /// penalty, and the recovery SPEC 8.2A requires would be unreachable for exactly the peer being
+    /// punished.
+    fn conduct_ticks(&self) -> u64 {
+        self.conduct_epoch.elapsed().as_secs()
+    }
+
     /// This node's routing memory, so a test can drive the forwarded ask and then ask what the ask
     /// LEFT BEHIND. Without it the recording leg would only be observable through its effect on a
     /// later round, and a test that could not see the write directly could not tell a missing write

crates/dig-node-core/src/lib.rs

@@ -492,6 +492,12 @@ pub struct Node {
     /// registry. The registry's distinct-generation cap ([`crate::seams::dig_peer::DEFAULT_MAX_CONCURRENT_WARMS`])
     /// therefore bounds concurrent acquisitions across BOTH legs, not each in isolation.
     capsule_acquisition: Arc<crate::seams::dig_peer::WarmRegistry>,
+    /// Inbound admission for the mTLS peer surface (dig-sex SPEC 8.5, #269).
+    ///
+    /// One meter for the whole NODE, not one per connection or per responder: the ceiling it enforces
+    /// is node-wide, and a per-connection meter would let a peer buy more allowance simply by opening
+    /// more connections.
+    peer_admission: Arc<crate::seams::dig_peer::admission::PeerAdmission>,
     /// A WEAK self-reference, installed by the standalone peer-network bring-up (which holds the
     /// `Arc<Node>`), so a `&self` read handler can spawn a detached background task that needs an owned
     /// `Arc<Node>` — the capsule backfill (§14.3). `Weak` (not `Arc`) so the node's refcount is
@@ -4171,7 +4177,11 @@ impl Node {
     /// a per-request walk of the whole cache is a cost amplifier a peer controls.
     ///
     /// The batch is CAPPED at [`MAX_AVAILABILITY_ITEMS`] — the item count is caller-controlled — with
-    /// the excess simply not answered (the result array is aligned to the answered prefix).
+    /// the excess simply not answered (the result array is aligned to the answered prefix). That
+    /// truncation is the LAST line of defence, reached only by in-process callers: on the peer
+    /// surface an oversized batch never gets here, because
+    /// `NodeResponder::handle_availability` (peer.rs) meters the requested item
+    /// count against the same limit and refuses the whole request at the boundary (#269).
     ///
     /// `requestor` keys the per-item not-held → DHT `find_providers` enrichment against its
     /// per-requestor miss-lookup budget (dig_ecosystem#2007), so a large batch of not-held items from
@@ -4676,6 +4686,11 @@ impl Node {
         self.capsule_acquisition.clone()
     }
 
+    /// The node-wide inbound admission meter (dig-sex SPEC 8.5, #269).
+    pub(crate) fn peer_admission(&self) -> &Arc<crate::seams::dig_peer::admission::PeerAdmission> {
+        &self.peer_admission
+    }
+
     /// Build a node from the environment (cache dir/cap, §21 identity, upstream).
     /// Used by both the standalone bin's [`run`] and the in-process `dig-runtime`.
     pub fn from_env() -> Arc<Node> {
@@ -4754,6 +4769,7 @@ impl Node {
             content_cache: std::sync::Mutex::new(ContentCache::default()),
             inventory_refresher: OnceLock::new(),
             capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()),
+            peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()),
             verification_ledger: verification_ledger::VerificationLedger::new(),
             self_ref: OnceLock::new(),
             gossip: OnceLock::new(),
@@ -5067,6 +5083,7 @@ pub(crate) mod test_support {
             content_cache: std::sync::Mutex::new(ContentCache::default()),
             inventory_refresher: OnceLock::new(),
             capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()),
+            peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()),
             verification_ledger: verification_ledger::VerificationLedger::new(),
             self_ref: OnceLock::new(),
             gossip: OnceLock::new(),
@@ -5856,6 +5873,7 @@ mod tests {
             content_cache: std::sync::Mutex::new(ContentCache::default()),
             inventory_refresher: OnceLock::new(),
             capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()),
+            peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()),
             verification_ledger: verification_ledger::VerificationLedger::new(),
             self_ref: OnceLock::new(),
             gossip: OnceLock::new(),
@@ -5989,6 +6007,7 @@ mod tests {
             content_cache: std::sync::Mutex::new(ContentCache::default()),
             inventory_refresher: OnceLock::new(),
             capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()),
+            peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()),
             verification_ledger: verification_ledger::VerificationLedger::new(),
             self_ref: OnceLock::new(),
             gossip: OnceLock::new(),
@@ -6056,6 +6075,7 @@ mod tests {
             content_cache: std::sync::Mutex::new(ContentCache::default()),
             inventory_refresher: OnceLock::new(),
             capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()),
+            peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()),
             verification_ledger: verification_ledger::VerificationLedger::new(),
             self_ref: OnceLock::new(),
             gossip: OnceLock::new(),
@@ -6148,6 +6168,9 @@ mod tests {
                 content_cache: std::sync::Mutex::new(ContentCache::default()),
                 inventory_refresher: OnceLock::new(),
                 capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()),
+                peer_admission: Arc::new(
+                    crate::seams::dig_peer::admission::PeerAdmission::default(),
+                ),
                 verification_ledger: verification_ledger::VerificationLedger::new(),
                 self_ref: OnceLock::new(),
                 gossip: OnceLock::new(),
@@ -6222,6 +6245,9 @@ mod tests {
                 content_cache: std::sync::Mutex::new(ContentCache::default()),
                 inventory_refresher: OnceLock::new(),
                 capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()),
+                peer_admission: Arc::new(
+                    crate::seams::dig_peer::admission::PeerAdmission::default(),
+                ),
                 verification_ledger: verification_ledger::VerificationLedger::new(),
                 self_ref: OnceLock::new(),
                 gossip: OnceLock::new(),
@@ -9208,6 +9234,7 @@ mod tests {
             content_cache: std::sync::Mutex::new(ContentCache::default()),
             inventory_refresher: OnceLock::new(),
             capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()),
+            peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()),
             verification_ledger: verification_ledger::VerificationLedger::new(),
             self_ref: OnceLock::new(),
             gossip: OnceLock::new(),
@@ -10988,6 +11015,10 @@ mod tests {
         assert_eq!(arr[2]["available"], false, "unknown capsule is a miss");
     }
 
+    /// Pins the IN-PROCESS truncation only. A peer-surface batch this size is refused whole at
+    /// admission long before it reaches here, and this test cannot see that: it calls the batch
+    /// BELOW the responder that decides. The responder-level pair lives in `peer.rs`
+    /// (`the_responder_serves_a_batch_at_the_advertised_limit_and_refuses_one_past_it`).
     #[tokio::test]
     async fn availability_batch_caps_the_item_count() {
         let (node, _td) = test_node(None);