feat(pointer): store, serve and pay for mutable references

Cargo.toml

@@ -228,6 +228,13 @@ webrtc-direct = [
     "dep:self_encryption",
 ]
 
+[patch.crates-io]
+# Pointers (ADR-0016) add the `Pointer` record and its wire messages on top of
+# the published 3.0.0, so this is the only entry that has to leave the release
+# baseline. A rev, not a branch, so the pin is immutable. Drop it once the
+# pointer PR lands and 3.1.0 is published.
+ant-protocol = { git = "https://github.com/WithAutonomi/ant-protocol", rev = "4412b6af4fab2ceb0a4e1efea1efca08e68c6e4e" }
+
 [profile.release]
 lto = true
 codegen-units = 1

docs/adr/README.md

@@ -35,3 +35,4 @@ See [`TOOLING.md`](./TOOLING.md) for `adrs`, `adr-kit`, and AI harness setup.
 - [ADR-0014: One File Per Chunk, and Retiring LMDB Without Losing Data](./ADR-0014-file-based-chunk-store-and-lmdb-retirement.md)
 - [ADR-0013: Settlement version and pre-payment compatibility](./ADR-0013-settlement-version-and-pre-payment-compatibility.md)
 - [ADR-0015: Direct browser clients over WebRTC Direct](./ADR-0015-direct-browser-clients-over-webrtc-direct.md)
+- [ADR-0016: Pointers — paid mutable references with an immutable owner](./ADR-0016-pointers-immutable-owner.md)

src/devnet.rs

@@ -973,11 +973,19 @@ impl Devnet {
         let storage = Arc::new(storage);
         let payment_verifier = Arc::new(payment_verifier);
 
-        Ok(AntProtocol::new(
-            storage,
-            payment_verifier,
-            Arc::new(quote_generator),
-        ))
+        // Same pointer wiring as a production node, so a devnet exercises the
+        // real path rather than a node that silently refuses every pointer.
+        let pointer_store = crate::pointer::PointerStore::new(storage.root_dir())
+            .await
+            .map_err(|e| DevnetError::Startup(format!("Failed to open pointer store: {e}")))?;
+        let pointers = crate::pointer::PointerService::new(pointer_store)
+            .with_chunk_store(Arc::clone(&storage))
+            .with_payments(Arc::clone(&payment_verifier));
+
+        Ok(
+            AntProtocol::new(storage, payment_verifier, Arc::new(quote_generator))
+                .with_pointer_service(pointers),
+        )
     }
 
     #[allow(clippy::too_many_lines)]

src/error.rs

@@ -60,3 +60,20 @@ pub enum Error {
     #[error("node is shutting down")]
     ShuttingDown,
 }
+
+impl From<ant_protocol::pointer::PointerError> for Error {
+    /// Map a wire-level pointer rejection onto the node's error type.
+    ///
+    /// Signature failures become [`Error::Crypto`] and everything else becomes
+    /// [`Error::Protocol`], so a caller can still tell "these bytes are not a
+    /// pointer" from "these bytes are not signed by the key they carry".
+    fn from(error: ant_protocol::pointer::PointerError) -> Self {
+        use ant_protocol::pointer::PointerError;
+        match error {
+            PointerError::SignatureInvalid | PointerError::SigningFailed(_) => {
+                Self::Crypto(error.to_string())
+            }
+            other => Self::Protocol(other.to_string()),
+        }
+    }
+}

src/lib.rs

@@ -17,8 +17,10 @@
 //!
 //! ## Data Types
 //!
-//! Currently supports a single data type:
+//! Two data types:
 //! - **Chunk**: Immutable content-addressed data (hash(value) == key)
+//! - **Pointer**: A paid mutable reference signed by an immutable owner, stored
+//!   at an address derived from the owner key (see [`mod@pointer`] and ADR-0016)
 //!
 //! ## Example
 //!
@@ -52,6 +54,7 @@ pub mod event;
 pub mod logging;
 pub mod node;
 pub mod payment;
+pub mod pointer;
 pub mod replication;
 pub mod storage;
 pub mod upgrade;
@@ -77,6 +80,9 @@ pub use error::{Error, Result};
 pub use event::{NodeEvent, NodeEventsChannel};
 pub use node::{NodeBuilder, RunningNode};
 pub use payment::{PaymentStatus, PaymentVerifier, PaymentVerifierConfig};
+pub use pointer::{
+    Pointer, PointerState, PointerStore, PointerTarget, PointerTargetKind, PutOutcome,
+};
 pub use replication::{config::ReplicationConfig, ReplicationEngine};
 pub use storage::{AntProtocol, ChunkStore, ChunkStoreConfig, LmdbStorage, LmdbStorageConfig};
 

src/node.rs

@@ -238,7 +238,7 @@ impl NodeBuilder {
         fresh_rx: UnboundedReceiver<FreshWriteEvent>,
         shutdown: &CancellationToken,
     ) -> Result<(Option<ReplicationEngine>, Option<JoinHandle<()>>)> {
-        let engine = match ReplicationEngine::new(
+        let mut engine = match ReplicationEngine::new(
             repl_config,
             Arc::clone(p2p),
             protocol.storage(),
@@ -271,6 +271,14 @@ impl NodeBuilder {
             }
         };
 
+        // ADR-0016: pointers replicate through the same engine. The PUT handler
+        // hands each newly stored paid state to it on this channel.
+        if let Some(service) = protocol.pointer_service() {
+            let (writes, fresh_writes) = tokio::sync::mpsc::unbounded_channel();
+            service.attach_fresh_writes(writes);
+            engine.with_pointers(service.store().clone(), fresh_writes);
+        }
+
         // ADR-0004: wire the engine's commitment state as the quote generator's
         // commitment source so quotes force their price from the live storage
         // commitment. Done here because the engine owns the commitment state and is
@@ -576,7 +584,21 @@ impl NodeBuilder {
         let storage = Arc::new(storage);
         let payment_verifier = Arc::new(payment_verifier);
 
-        let protocol = AntProtocol::new(storage, payment_verifier, Arc::new(quote_generator));
+        // Pointers live beside the chunks, under the same root. Opening the
+        // store here is what makes pointer PUT/GET answerable at all: without
+        // it every pointer request is refused, which is the right answer for a
+        // node that keeps none but the wrong one for a node that should.
+        let pointer_store = crate::pointer::PointerStore::new(&config.root_dir).await?;
+        let pointers = crate::pointer::PointerService::new(pointer_store)
+            // Refuse a pointer whose address a chunk already occupies, rather
+            // than letting one kind silently overwrite the other.
+            .with_chunk_store(Arc::clone(&storage))
+            // Payment is verified against each record's state, so every update
+            // is paid for rather than riding the first one.
+            .with_payments(Arc::clone(&payment_verifier));
+
+        let protocol = AntProtocol::new(storage, payment_verifier, Arc::new(quote_generator))
+            .with_pointer_service(pointers);
 
         info!(
             "ANT protocol handler initialized with ML-DSA-65 signing (protocol={CHUNK_PROTOCOL_ID})"

src/payment/cache.rs

@@ -10,6 +10,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 
 pub use super::quote::XorName;
+pub use super::verifier::PaymentTarget as PaidKey;
 
 /// Default cache capacity (100,000 entries = 3.2MB memory).
 const DEFAULT_CACHE_CAPACITY: usize = 100_000;
@@ -25,7 +26,7 @@ const DEFAULT_CACHE_CAPACITY: usize = 100_000;
 /// entries satisfy weaker lookups.
 #[derive(Clone)]
 pub struct VerifiedCache {
-    inner: Arc<Mutex<LruCache<XorName, VerificationLevel>>>,
+    inner: Arc<Mutex<LruCache<PaidKey, VerificationLevel>>>,
     hits: Arc<AtomicU64>,
     misses: Arc<AtomicU64>,
     additions: Arc<AtomicU64>,
@@ -101,8 +102,8 @@ impl VerifiedCache {
     /// Returns `true` if the `XorName` is cached (verified to exist on autonomi).
     /// Paid-list and client-PUT lookups must use their stricter helpers.
     #[must_use]
-    pub fn contains(&self, xorname: &XorName) -> bool {
-        let found = self.inner.lock().get(xorname).is_some();
+    pub fn contains_key(&self, key: &PaidKey) -> bool {
+        let found = self.inner.lock().get(key).is_some();
 
         if found {
             self.hits.fetch_add(1, Ordering::Relaxed);
@@ -119,11 +120,11 @@ impl VerifiedCache {
     /// A client-PUT entry returns `true` here because it passed the stricter
     /// store-admission path at the caller.
     #[must_use]
-    pub fn contains_paid_list_verified(&self, xorname: &XorName) -> bool {
+    pub fn contains_paid_list_verified_key(&self, key: &PaidKey) -> bool {
         let found = self
             .inner
             .lock()
-            .get(xorname)
+            .get(key)
             .copied()
             .is_some_and(|level| level.satisfies(VerificationLevel::PaidList));
 
@@ -142,11 +143,11 @@ impl VerifiedCache {
     /// Paid-list entries return `false` here because they did not pass the
     /// client-PUT store-admission path.
     #[must_use]
-    pub fn contains_client_put_verified(&self, xorname: &XorName) -> bool {
+    pub fn contains_client_put_verified_key(&self, key: &PaidKey) -> bool {
         let found = self
             .inner
             .lock()
-            .get(xorname)
+            .get(key)
             .copied()
             .is_some_and(|level| level.satisfies(VerificationLevel::ClientPut));
 
@@ -163,30 +164,30 @@ impl VerifiedCache {
     ///
     /// This should be called after verifying that data exists on the autonomi network.
     /// Also upgrades an existing paid-list-verified entry.
-    pub fn insert(&self, xorname: XorName) {
-        self.insert_with_level(xorname, VerificationLevel::ClientPut);
+    pub fn insert_key(&self, key: PaidKey) {
+        self.insert_with_level(key, VerificationLevel::ClientPut);
     }
 
     /// Add a `XorName` verified under paid-list admission checks.
     ///
     /// Never downgrades an existing client-PUT-verified entry.
-    pub fn insert_paid_list_verified(&self, xorname: XorName) {
-        self.insert_with_level(xorname, VerificationLevel::PaidList);
+    pub fn insert_paid_list_verified_key(&self, key: PaidKey) {
+        self.insert_with_level(key, VerificationLevel::PaidList);
     }
 
-    fn insert_with_level(&self, xorname: XorName, level: VerificationLevel) {
+    fn insert_with_level(&self, key: PaidKey, level: VerificationLevel) {
         let added = {
             let mut inner = self.inner.lock();
             // `get_mut` refreshes LRU recency for existing entries of either kind.
-            if inner.get(&xorname).is_some() {
-                if let Some(existing) = inner.get_mut(&xorname) {
+            if inner.get(&key).is_some() {
+                if let Some(existing) = inner.get_mut(&key) {
                     if !existing.satisfies(level) {
                         *existing = level;
                     }
                 }
                 false
             } else {
-                inner.put(xorname, level);
+                inner.put(key, level);
                 true
             }
         };
@@ -195,6 +196,34 @@ impl VerifiedCache {
         }
     }
 
+    /// As [`Self::contains_key`], for a chunk at `address`.
+    #[must_use]
+    pub fn contains(&self, address: &XorName) -> bool {
+        self.contains_key(&PaidKey::Chunk(*address))
+    }
+
+    /// As [`Self::contains_paid_list_verified_key`], for a chunk at `address`.
+    #[must_use]
+    pub fn contains_paid_list_verified(&self, address: &XorName) -> bool {
+        self.contains_paid_list_verified_key(&PaidKey::Chunk(*address))
+    }
+
+    /// As [`Self::contains_client_put_verified_key`], for a chunk at `address`.
+    #[must_use]
+    pub fn contains_client_put_verified(&self, address: &XorName) -> bool {
+        self.contains_client_put_verified_key(&PaidKey::Chunk(*address))
+    }
+
+    /// As [`Self::insert_key`], for a chunk at `address`.
+    pub fn insert(&self, address: XorName) {
+        self.insert_key(PaidKey::Chunk(address));
+    }
+
+    /// As [`Self::insert_paid_list_verified_key`], for a chunk at `address`.
+    pub fn insert_paid_list_verified(&self, address: XorName) {
+        self.insert_paid_list_verified_key(PaidKey::Chunk(address));
+    }
+
     /// Get current cache statistics.
     #[must_use]
     pub fn stats(&self) -> CacheStats {
@@ -238,23 +267,23 @@ mod tests {
     fn test_cache_basic_operations() {
         let cache = VerifiedCache::new();
 
-        let xorname1 = [1u8; 32];
-        let xorname2 = [2u8; 32];
+        let key1 = [1u8; 32];
+        let key2 = [2u8; 32];
 
         // Initially empty
         assert!(cache.is_empty());
-        assert!(!cache.contains(&xorname1));
+        assert!(!cache.contains(&key1));
 
         // Insert and check
-        cache.insert(xorname1);
-        assert!(cache.contains(&xorname1));
-        assert!(!cache.contains(&xorname2));
+        cache.insert(key1);
+        assert!(cache.contains(&key1));
+        assert!(!cache.contains(&key2));
         assert_eq!(cache.len(), 1);
 
         // Insert another
-        cache.insert(xorname2);
-        assert!(cache.contains(&xorname1));
-        assert!(cache.contains(&xorname2));
+        cache.insert(key2);
+        assert!(cache.contains(&key1));
+        assert!(cache.contains(&key2));
         assert_eq!(cache.len(), 2);
     }
 
@@ -284,21 +313,21 @@ mod tests {
     #[test]
     fn test_cache_stats() {
         let cache = VerifiedCache::new();
-        let xorname = [1u8; 32];
+        let key = [1u8; 32];
 
         // Miss
-        assert!(!cache.contains(&xorname));
+        assert!(!cache.contains(&key));
         let stats = cache.stats();
         assert_eq!(stats.misses, 1);
         assert_eq!(stats.hits, 0);
 
         // Add
-        cache.insert(xorname);
+        cache.insert(key);
         let stats = cache.stats();
         assert_eq!(stats.additions, 1);
 
         // Hit
-        assert!(cache.contains(&xorname));
+        assert!(cache.contains(&key));
         let stats = cache.stats();
         assert_eq!(stats.hits, 1);
         assert_eq!(stats.misses, 1);
@@ -312,19 +341,19 @@ mod tests {
         // Small cache for testing eviction
         let cache = VerifiedCache::with_capacity(2);
 
-        let xorname1 = [1u8; 32];
-        let xorname2 = [2u8; 32];
-        let xorname3 = [3u8; 32];
+        let key1 = [1u8; 32];
+        let key2 = [2u8; 32];
+        let key3 = [3u8; 32];
 
-        cache.insert(xorname1);
-        cache.insert(xorname2);
+        cache.insert(key1);
+        cache.insert(key2);
         assert_eq!(cache.len(), 2);
 
-        // Insert third, should evict xorname1 (least recently used)
-        cache.insert(xorname3);
+        // Insert third, should evict key1 (least recently used)
+        cache.insert(key3);
         assert_eq!(cache.len(), 2);
-        assert!(!cache.contains(&xorname1)); // evicted
-                                             // Note: after contains call on evicted item, stats will show a miss
+        assert!(!cache.contains(&key1)); // evicted
+                                         // Note: after contains call on evicted item, stats will show a miss
     }
 
     #[test]
@@ -409,17 +438,17 @@ mod tests {
         for i in 0..10u8 {
             let c = cache.clone();
             handles.push(thread::spawn(move || {
-                let xorname = [i; 32];
-                c.insert(xorname);
+                let key = [i; 32];
+                c.insert(key);
             }));
         }
 
         // 10 threads checking
         for i in 0..10u8 {
             let c = cache.clone();
             handles.push(thread::spawn(move || {
-                let xorname = [i; 32];
-                let _ = c.contains(&xorname);
+                let key = [i; 32];
+                let _ = c.contains(&key);
             }));
         }