# Cluster-Wide Registry — Distributed Naming

## Problem

Actors can only be found by their `ActorAddress` (a random 32-byte ID). The local `AddressMap` maps addresses to workers on a single node. The Kademlia directory maps addresses to `NodeId`. But neither provides **human-readable naming** or **re-discovery after churn**.

When a node dies and an actor is re-spawned elsewhere, it gets a new `ActorAddress`. Without a name-based registry, every actor that communicated with it needs manual reconfiguration. This doesn't work for churning infrastructure.

## Design

### Approach: Gossip-Propagated LWW-Register CRDT

Each name binding is a **Last-Writer-Wins Register** — the most recent write (by timestamp) wins. This matches SWIM's eventual-consistency model and reuses the existing gossip piggyback mechanism.

**Why not Raft/consensus?**
- Overkill for name resolution. Names don't need linearizability — eventual consistency is fine.
- SWIM already solves dissemination. We piggyback registry updates on existing protocol messages for free.
- Consensus requires a stable quorum, which conflicts with the "nodes pop in and out" use case.

**Why not extend Kademlia?**
- Kademlia maps `ActorAddress -> NodeId`. Names are a different key space (`String -> ActorAddress`).
- Kademlia lookups are multi-hop (iterative). Registry lookups should be local (every node has a full replica).
- The registry is small (hundreds to low-thousands of names). Full replication is cheap.

### Types

```rust
// crates/distribution/src/registry.rs

/// A single name binding in the cluster registry.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RegistryEntry {
    /// Human-readable name (e.g. "worker-pool", "metrics-collector").
    pub name: String,
    /// The actor address this name resolves to.
    pub actor_addr: ActorAddress,
    /// The node that owns this binding.
    pub node_id: NodeId,
    /// Logical timestamp for LWW conflict resolution.
    pub timestamp: u64,
    /// Generation — incremented on re-registration of the same name.
    pub generation: u64,
    /// Tombstone — true means the name has been unregistered.
    pub tombstone: bool,
}

/// Events emitted by the registry for subscribers.
#[derive(Debug, Clone)]
pub enum RegistryEvent {
    /// A name was registered or updated.
    Registered {
        name: String,
        actor_addr: ActorAddress,
        node_id: NodeId,
    },
    /// A name was unregistered (tombstoned).
    Unregistered {
        name: String,
        previous_addr: ActorAddress,
    },
}

/// The local replica of the cluster-wide registry.
pub struct ClusterRegistry {
    /// Current state: name -> latest entry.
    entries: HashMap<String, RegistryEntry>,
    /// Pending entries to propagate via gossip (not yet disseminated to all).
    pending: VecDeque<RegistryEntry>,
    /// Logical clock for this node.
    clock: u64,
    /// Recent events for subscribers.
    events: VecDeque<RegistryEvent>,
    /// Max events to buffer.
    max_events: usize,
}
```

### CRDT Merge Rule

```rust
impl ClusterRegistry {
    /// Merge a remote entry. Returns true if the local state changed.
    pub fn merge(&mut self, remote: RegistryEntry) -> bool {
        match self.entries.get(&remote.name) {
            Some(local) => {
                // LWW: higher timestamp wins.
                // Tie-break: higher generation, then higher node_id (deterministic).
                let dominated = remote.timestamp > local.timestamp
                    || (remote.timestamp == local.timestamp
                        && remote.generation > local.generation)
                    || (remote.timestamp == local.timestamp
                        && remote.generation == local.generation
                        && remote.node_id.0 > local.node_id.0);

                if dominated {
                    self.apply(remote);
                    true
                } else {
                    false
                }
            }
            None => {
                self.apply(remote);
                true
            }
        }
    }

    fn apply(&mut self, entry: RegistryEntry) {
        let event = if entry.tombstone {
            let prev = self.entries.get(&entry.name)
                .map(|e| e.actor_addr);
            RegistryEvent::Unregistered {
                name: entry.name.clone(),
                previous_addr: prev.unwrap_or_default(),
            }
        } else {
            RegistryEvent::Registered {
                name: entry.name.clone(),
                actor_addr: entry.actor_addr,
                node_id: entry.node_id,
            }
        };
        self.events.push_back(event);
        if self.events.len() > self.max_events {
            self.events.pop_front();
        }
        self.entries.insert(entry.name.clone(), entry);
    }
}
```

### API

On `DistributedNode`:

```rust
// crates/distribution/src/node.rs

impl DistributedNode {
    /// Register a name -> actor binding on this node.
    /// The binding is propagated to all cluster members via gossip.
    pub fn register_name(&mut self, name: &str, actor_addr: ActorAddress) {
        self.registry.clock += 1;
        let entry = RegistryEntry {
            name: name.to_string(),
            actor_addr,
            node_id: self.node_id(),
            timestamp: self.registry.clock,
            generation: self.registry.next_generation(name),
            tombstone: false,
        };
        self.registry.merge(entry.clone());
        self.registry.pending.push_back(entry);
    }

    /// Remove a name binding. Propagated as a tombstone.
    pub fn unregister_name(&mut self, name: &str) {
        self.registry.clock += 1;
        let actor_addr = self.registry.entries.get(name)
            .map(|e| e.actor_addr)
            .unwrap_or_default();
        let entry = RegistryEntry {
            name: name.to_string(),
            actor_addr,
            node_id: self.node_id(),
            timestamp: self.registry.clock,
            generation: 0,
            tombstone: true,
        };
        self.registry.merge(entry.clone());
        self.registry.pending.push_back(entry);
    }

    /// Resolve a name to an actor address (local replica, eventually consistent).
    pub fn resolve_name(&self, name: &str) -> Option<(ActorAddress, NodeId)> {
        self.registry.entries.get(name)
            .filter(|e| !e.tombstone)
            .map(|e| (e.actor_addr, e.node_id))
    }

    /// Drain buffered registry events (for subscribers).
    pub fn registry_events(&mut self) -> Vec<RegistryEvent> {
        self.registry.events.drain(..).collect()
    }
}
```

On `Ctx` (actor-level, requires distribution feature):

```rust
// src/actor.rs — requires ContextInner extensions

impl Ctx<'_> {
    /// Register this actor under a name in the cluster registry.
    pub fn register_as(&self, name: &str) {
        self.inner.register_name(self.self_addr, name);
    }

    /// Resolve a name to an actor address.
    pub fn resolve_name(&self, name: &str) -> Option<ActorAddress> {
        self.inner.resolve_name(name)
    }
}
```

The `ContextInner` trait gains two new methods:

```rust
pub trait ContextInner {
    // ... existing methods ...
    fn register_name(&self, addr: ActorAddress, name: &str) { /* default no-op */ }
    fn resolve_name(&self, name: &str) -> Option<ActorAddress> { None }
}
```

Default implementations return `None` / no-op so that non-distributed runtimes don't break.

### Gossip Propagation

Registry entries are piggybacked on SWIM protocol messages, reusing the existing dissemination mechanism.

Currently, `crates/distribution/src/swim/dissemination.rs` encodes membership updates into the piggyback payload:

```
piggyback bytes = bincode(Vec<MembershipUpdate>)
```

Extended format:

```
piggyback bytes = bincode(PiggybackPayload {
    membership: Vec<MembershipUpdate>,
    registry: Vec<RegistryEntry>,       // NEW
})
```

```rust
// crates/distribution/src/swim/dissemination.rs

#[derive(Serialize, Deserialize)]
struct PiggybackPayload {
    membership: Vec<MembershipUpdate>,
    registry: Vec<RegistryEntry>,
}
```

The dissemination buffer manages registry entries the same way as membership updates:
- Each entry has a dissemination count (how many times it's been piggybacked).
- After `log2(N) + 1` disseminations (where N = cluster size), the entry is retired.
- Piggyback space is shared: membership updates take priority, registry entries fill remaining space.

### Node Death Handling

When SWIM marks a node as `Dead`:

```rust
fn handle_membership_change(&mut self, node_id: NodeId, state: MemberState) {
    if state == MemberState::Dead {
        // ... existing cleanup ...

        // NEW: tombstone all registry entries owned by the dead node
        let to_tombstone: Vec<String> = self.registry.entries.iter()
            .filter(|(_, e)| e.node_id == node_id && !e.tombstone)
            .map(|(name, _)| name.clone())
            .collect();

        for name in to_tombstone {
            self.registry.clock += 1;
            let entry = RegistryEntry {
                name: name.clone(),
                tombstone: true,
                timestamp: self.registry.clock,
                // ... fill from existing entry ...
            };
            self.registry.merge(entry.clone());
            self.registry.pending.push_back(entry);
        }
    }
}
```

### Interaction with Actor Watching

The registry and watching system compose naturally:

1. Actor A resolves name "service-X" → gets address B on Node 2.
2. Actor A calls `ctx.watch(B)`.
3. Node 2 dies. Actor A receives `ActorExited { addr: B, reason: NodeDown }`.
4. A supervisor re-spawns "service-X" on Node 3 → new address C.
5. The supervisor calls `register_name("service-X", C)`.
6. Gossip propagates the update.
7. Actor A (or anyone) calls `resolve_name("service-X")` → gets address C.
8. Actor A calls `ctx.watch(C)` to resume monitoring.

### Tombstone Garbage Collection

Tombstones accumulate over time. GC strategy:

- Tombstones older than `tombstone_ttl` (default: 1 hour of logical clock ticks) are eligible for removal.
- GC runs periodically (e.g., every 1000 ticks).
- A tombstone is only removed if it has been fully disseminated (dissemination count >= threshold).

### Files Modified

| File | Change |
|------|--------|
| `crates/distribution/src/registry.rs` | **New file**: `ClusterRegistry`, `RegistryEntry`, `RegistryEvent`, CRDT merge |
| `crates/distribution/src/lib.rs` | `pub mod registry;` |
| `crates/distribution/src/node.rs` | `register_name`, `unregister_name`, `resolve_name`, node death tombstoning |
| `crates/distribution/src/swim/dissemination.rs` | `PiggybackPayload` extended with registry entries |
| `src/actor.rs` | `register_name`/`resolve_name` on `ContextInner` (default no-op), `Ctx` wrappers |

### Tests

- **register_and_resolve**: register a name, resolve it, verify correct address
- **lww_conflict**: two nodes register same name concurrently, verify latest timestamp wins
- **tombstone_propagation**: register name, unregister, verify tombstone propagates and resolve returns None
- **node_death_tombstones**: 3-node cluster, register name on node B, kill node B, verify name is tombstoned on surviving nodes
- **re_registration**: register name, unregister, re-register with new address, verify resolution
- **gossip_convergence**: register name on node A, verify all nodes resolve it after gossip settles