Compare commits

..
18 Commits
Author SHA1 Message Date
nils 3e39f34383 feat: integrate iroh-gossip for mesh networking, using ALPN for protocol routing and node events to manage gossip topics. 2025-12-23 02:51:28 +01:00
nils 4ccdbc97f5 refactor: Consolidate network sync operations into LatticeServer methods 2025-12-23 01:41:34 +01:00
nils 2de54d0033 feat: make open_store async and add root store caching 2025-12-23 00:48:55 +01:00
nils 9d4495b3d7 feat: Refactor mesh server to use Arc<Node>, introduce JoinAcceptance for mesh joins, and enhance store listing with prefix filtering and deleted entry inclusion. 2025-12-23 00:31:54 +01:00
nils 76810f8d8e feat: Refactor CLI commands into dedicated modules, update roadmap, and simplify sync function signature. 2025-12-22 23:06:36 +01:00
nils 57ecbffaed feat: introduce NodeIdentity and store_actor in lattice-core, and implement mesh networking in lattice-net while removing unicast. 2025-12-22 22:11:20 +01:00
nils 665114036b refactor: remove lattice-store crate and update roadmap tasks 2025-12-22 21:12:29 +01:00
nils e942da49ff feat: Implement Iroh-based peer networking, join protocol, and bidirectional store synchronization. 2025-12-22 21:11:34 +01:00
nils 7c8e5cfa3d feat: Record node metadata including public key, hostname, and status in the root store during initialization. 2025-12-22 03:58:49 +01:00
nils 1943e06509 feat: Implement new synchronization state management and deterministic head sorting, replacing the old vector clock module. 2025-12-22 03:42:15 +01:00
nils 4761501ec9 feat: Migrate CLI store actor and handle to tokio async channels and operations 2025-12-22 03:16:26 +01:00
nils c2d4219320 feat: Refactor node initialization and info retrieval, enhance store actor lifecycle, and refine store replay logic. 2025-12-22 02:59:55 +01:00
nils a1f134eb02 feat: implement idempotent put and delete operations by checking existing heads before committing new entries 2025-12-22 02:28:36 +01:00
nils ce852da25e feat: implement store actor pattern for CLI store operations and update roadmap 2025-12-22 02:13:08 +01:00
nils 57c2906b10 feat: Implement DAG-based conflict resolution with binary keys and multi-head CLI display 2025-12-22 01:56:26 +01:00
nils 346ebccee7 feat: introduce global meta store and root store concept, and update CLI to manage active store 2025-12-21 23:59:08 +01:00
nils f45c6ccfcf feat: implement interactive CLI with key-value operations, add development journal, and update roadmap for DAG conflict resolution 2025-12-21 22:59:58 +01:00
nils 15dd1b337f feat: Add Store module for persistent KV state, update lib.rs and documentation including a development journal and architecture details. 2025-12-21 21:56:05 +01:00
38 changed files with 6275 additions and 416 deletions
+11 -3
View File
@@ -3,7 +3,7 @@ resolver = "2"
members = [
"lattice-core",
"lattice-net",
"lattice-store",
"lattice-cli",
]
[workspace.package]
@@ -15,10 +15,13 @@ license = "MIT"
# Workspace crates
lattice-core = { path = "lattice-core" }
lattice-net = { path = "lattice-net" }
lattice-store = { path = "lattice-store" }
lattice-cli = { path = "lattice-cli" }
# CLI
rustyline = "17"
# Networking (Iroh)
iroh = "0.95"
iroh = { version = "0.95", features = ["discovery-local-network"] }
iroh-gossip = "0.95"
# Cryptography
@@ -32,6 +35,8 @@ prost-build = "0.13"
# Async runtime
tokio = { version = "1", features = ["full"] }
tokio-util = { version = "0.7", features = ["codec"] }
futures-util = "0.3"
# Utilities
thiserror = "2"
@@ -40,6 +45,9 @@ bytes = "1"
dirs = "5"
blake3 = "1"
hex = "0.4"
redb = "2"
uuid = { version = "1", features = ["v4"] }
chrono = "0.4"
# Testing
tokio-test = "0.4"
+181 -24
View File
@@ -2,20 +2,38 @@
## Ideas
- SigChains Ed25519-signed, hash-chained append-only logs per node. Trust via local signature verification.
- Log-Based State: KV store derived by replaying entries. Watermarks enable safe log pruning + snapshots.
- Offline-First: Iroh for networking. Vector clocks identify missing entries on reconnect—converges mathematically.
- Full Replication: All nodes keep all logs until watermark consensus, then prune and snapshot.
**Core:**
- SigChains: Ed25519-signed, hash-chained append-only logs per node.
- Offline-First: Iroh for networking. Vector clocks identify missing entries on reconnect.
- Full Replication: All nodes keep all logs until watermark consensus, then prune.
**State:**
- Log-Based State: KV store derived from entries. Watermarks enable pruning + snapshots.
- Merkle-ized State: state.db as Merkle tree. O(1) sync checks, efficient diffing, light clients.
- DAG Conflict Resolution: Entries track ancestry. Forks merge on next write. Tips only in state.db.
- KV Snapshots: Point-in-time snapshots for log pruning, fast bootstrap, time travel.
**Operations:**
- Atomic Batch Writes: Multiple key updates as single entry.
- Conditional Updates (CAS): Update only if current value matches expected hash.
**CRDTs:**
- LWW-Register: Last-writer-wins for single values.
- LWW-Element-Set: Set with add/remove, element present if add > remove timestamp.
## Concepts
- Transitive Pairing. Nodes can introduce new nodes to the mesh.
- Transitive Pairing: Nodes can introduce new nodes to the mesh.
- Multi-Mesh: A node can participate in multiple meshes (clusters). Each mesh is a group of nodes sharing data.
- Manifest Store: Joining a mesh means joining a special KV store of type "manifest" that defines the mesh membership. The manifest contains node info (`/nodes/{pubkey}/...`).
## Stack
- rust
- iroh
- prost protocol buffers
- redb (embedded KV store)
- rustyline (interactive CLI)
### Bootstrap
@@ -41,10 +59,17 @@ Networking modes:
- Identified by their Ed25519 public key.
- Private key stored locally in `identity.key` (not replicated).
- Node data stored in KV:
- `/nodes/{pubkey}/info` = static metadata (name, added_by, added_at)
- `/nodes/{pubkey}/status` = `active` | `dormant` | `disabled`
- `/nodes/{pubkey}/name` = display name
- `/nodes/{pubkey}/added_at` = timestamp when added
- `/nodes/{pubkey}/status` = `invited` | `active` | `dormant` (removal deletes keys)
- `/nodes/{pubkey}/role` = `server` | `device` (optional, hints sync priority)
- Inviting a node = writing entries to `/nodes/{pubkey}/...`.
- Peer invitation flow:
1. Inviter runs `invite <peer_pubkey>` → writes `/nodes/{peer}/info` + `/status`
2. Inviter shares their Iroh NodeId out-of-band (QR code, link, text)
3. Invited peer runs `join <inviter_nodeid>` → syncs with inviter
4. Sync pulls `/nodes/{self}/info` + `/status` → peer is authorized
5. `connect` implicitly adds inviter to peer's `/nodes/*` (mutual awareness)
- Accepting = syncing. The invited peer discovers authorization by receiving the entries.
- Liveness: Each node tracks `last_seen` locally (from watermark gossip). UI alerts if a peer hasn't been seen for threshold (e.g., 30 days). User decides to mark dormant/disabled.
- Status effects:
- `active`: Normal sync participant, blocks watermark until acknowledged.
@@ -58,11 +83,62 @@ Future:
### Data Model
- Keys are flat strings using path conventions (e.g., `/nodes/{pubkey}`, `/config/sync/interval`).
- Prefix queries via string matching (sorted map enables efficient range scans).
- State is computed by replaying `Put`/`Delete` operations from all authors.
- Multiple KV stores supported, identified by `store_id` (UUID).
- Keys: Arbitrary byte arrays (`Vec<u8>`), sorted lexicographically.
- Values: Arbitrary byte arrays (`Vec<u8>`).
- Each store defines its own key/value format — applications know their schema.
- Logs are per `(store_id, author_id)` tuple.
- State is maintained by tracking the "frontier" (tips) of the causal graph for each key.
- Entry ordering: by HLC timestamp, then by author ID as tiebreaker.
- Conflicts resolved by last-write-wins (using the ordering above).
**Sync vs Causality:**
- Vector Clocks track log coverage ("I have entries from Node A up to seq 50") — syncing files.
- DAG Parents track data causality ("This value replaces that value") — resolving key conflicts.
#### DAG Conflict Resolution
Instead of simple LWW where newest timestamp blindly overwrites, every entry tracks its ancestry:
**Data Model:**
- Each entry includes `parent_hashes` — references to the entries it supersedes
- History forms a DAG (directed acyclic graph), not a linear chain
- state.db stores only "tips" (heads) of the graph per key
**Life Cycle:**
1. **Write (normal):** New entry points to previous entry's hash as parent. History is a straight line.
2. **Write (concurrent/offline):** Two nodes edit same key independently, both pointing to same old parent. History forks into two branches.
3. **Read (forked):** System sees multiple valid values. Uses deterministic rule (highest HLC, then author_id tiebreaker) to return one "winner". No error thrown.
4. **Merge (healing):** Next write to that key cites both existing branches as parents. Fork merges back to single tip.
**Example: Partial Write (Branch Extension)**
```
Initial: Heads = {A, B} where A(ts:100), B(ts:105). Read winner = B.
Offline node C wakes up, only knows A (hasn't seen B).
C writes "v3" with parent = [A].
Result: Heads = {C, B}. Conflict shifted, not resolved.
C(ts:110) > B(ts:105), so C wins reads.
┌──> [A] ──> [C:110]
[Root]─┤
└──> [B:105]
Later: A synced node writes D with parents = [C, B].
Result: Heads = {D}. Fork merged.
```
This preserves B's work even though C never saw it. Naive LWW would lose B forever.
#### Store Consistency Modes
- **Eventually consistent**: Default. Writes accepted locally, sync happens async. Fast, offline-capable.
- **Strictly consistent**: Writes require quorum acknowledgment before commit. Slower, requires connectivity.
### Timestamps (Hybrid Logical Clocks)
@@ -86,27 +162,80 @@ Authors apply their own entries through the standard receive path to ensure cons
Each node stores logs as one file per author:
```
data/
├── identity.key # Local node's Ed25519 private key
├── logs/
│ └── {author_id_hex}.log # Append-only SignedEntry stream per author
└── state.db # redb: KV snapshot + vector clocks + indexes
~/.local/share/lattice/
├── identity.key # Ed25519 private key
├── stores/
│ └── {store_uuid}/
│ ├── logs/
│ │ └── {author_id_hex}.log # Append-only SignedEntry stream
│ └── state.db # redb: KV snapshot + frontiers
└── meta.db # redb: global metadata (known stores, peers)
```
- Logs: Append-only binary files per author, containing serialized `SignedEntry` messages.
- State DB (redb): Combined KV state, vector clocks, and indexes. Updated as entries are applied.
- Logs: Append-only binary files per `(store, author)`, containing serialized `SignedEntry` messages.
- State DB (redb): Per-store KV state and frontiers. Updated as entries are applied.
#### state.db Tables (redb)
#### state.db Tables (per store, redb)
```
Table Key Value Purpose
─────────────────────────────────────────────────────────────────────────────
kv String (path) Vec<u8> Replicated key-value data
vector_clocks [u8; 32] (author_id) (u64 seq, [u8; 32] hash) Track sync state + chain verification
entry_index (author_id, seq) u64 (offset) Fast entry lookup by position
meta String Vec<u8> System metadata (own_seq, watermark, etc.)
kv Vec<u8> (key) Vec<HeadInfo> Current tips for each key
AUTHOR_TABLE [u8; 32] (author_id) (u64 seq, [u8; 32] hash) Per-author frontier tracking
meta Vec<u8> Vec<u8> Store metadata (incl. merkle_root)
```
`HeadInfo: { value: Vec<u8>, hlc: u64, author: [u8;32], hash: [u8;32] }`
Note: KV stores multiple heads per key to support DAG conflict resolution. Reads pick winner deterministically.
#### meta.db Tables (global, redb)
```
Table Key Value Purpose
─────────────────────────────────────────────────────────────────────────────
stores [u8; 16] (UUID) u64 (created_at_ms) Known stores
meta "root_store" [u8; 16] (UUID) Root store ID (opened on startup)
```
- **Root Store**: The primary/manifest store for this node, auto-opened on CLI startup
- **Stores Table**: Tracks all stores this node participates in
- Manifest stores define mesh membership via `/nodes/{pubkey}/...` entries
- Data stores hold application data
#### In-Memory Structures
- log_frontiers: `HashMap<AuthorId, (seq, hash)>` — rebuilt from log files on startup
### Operation Flow (put/delete)
```
1. User calls put("/key", value)
2. SigChain.create_entry()
- Build Entry with parent_hashes (current tips for key)
- Sign it → SignedEntry
3. Append to log + Gossip (critical path)
- Write to author's log file
- Update log_frontiers (in-memory)
- Broadcast to peers
4. Apply to state.db (background)
- Update kv heads (merge parent tips into new tip)
- Update applied_frontiers
- Update merkle_root hash
```
Fast path (1-3): durable + distributed. Background (4): queryable state.
### Read Flow (get)
`get(key)` reads directly from local state.db. Reads are eventually consistent — if state.db lags behind the log, the read may return slightly stale data.
### Watermarks
- Nodes gossip their watermarks periodically (throttled).
@@ -114,6 +243,11 @@ meta String Vec<u8> System metada
- All nodes keep all logs (own + others) for redundancy until watermark consensus.
- Once all peers have acknowledged entries, they can be pruned and replaced by the snapshot.
- If a node is offline too long, it re-bootstraps with a fresh snapshot when it reconnects.
- Note: Consider preserving logs longer than required for redundancy — enables time travel (view state at any point in history).
**Pruning and DAG Parents:**
- If a new entry references a parent that was pruned, accept it only if strictly newer than snapshot timestamp.
- Snapshots act as the base; entries referencing parents older than snapshot are roots relative to that snapshot.
### Rich CRDTs (Future)
@@ -157,3 +291,26 @@ message MergeOp {
```
Recommendation: Use Put/Delete for 90% of data. Add CRDT primitives only when needed (concurrent counters, lists) rather than a scripting language.
## Open Questions
### Permissions
Write permissions are enforceable cryptographically:
- Every entry is signed by author
- Nodes verify signature before accepting
- Manifest defines allowed writers: `/nodes/{pubkey}/role` = `writer` | `reader`
- Entries from non-writers are rejected
Read permissions are not enforceable:
- Sharing a store = granting read access
- Encryption adds a layer but doesn't solve revocation (once you have the key, you can read past data)
- True revocation is impossible — you can't "unread" data
Practical model:
- Share store = grant read
- Write access defined in manifest
- Read-only nodes replicate and verify but can't contribute entries
Future:
- Capability-based permissions: Explore finer-grained write access (e.g., per-key or per-prefix permissions) via capabilities. Exact mechanism TBD.
+161 -20
View File
@@ -12,7 +12,7 @@
- [x] Log file I/O (append, read, hash verification)
- [x] SigChain (validate entries before appending)
- [x] Store (redb) — `kv` + `meta` tables, log replay
- [ ] Interactive CLI: `init`, `put`, `get`, `delete`, `status`, `quit`
- [x] Interactive CLI: `init`, `put`, `get`, `delete`, `status`, `quit`
### Success Criteria
@@ -21,15 +21,63 @@
- Can replay log to reconstruct KV state
- All operations survive restart
### Multi-KV Refactoring (before M2)
### Multi-KV Refactoring (before M2)
Current code assumes single store. Changes needed:
- [ ] DataDir → support `stores/{uuid}/` subdirectories
- [ ] SigChain → scoped to (store_id, author_id)
- [ ] Store → per-store state.db, not global
- [ ] Log paths → `stores/{uuid}/logs/{author}.log`
- [ ] Add global meta.db for stores table
- [ ] CLI → `create-store`, `list-stores`, `use <store>`
- [x] DataDir → `stores/{uuid}/` subdirectories
- [x] Store → per-store state.db
- [x] Log paths → `stores/{uuid}/logs/{author}.log`
- [x] Proto: Entry has store_id (UUID)
- [x] CLI → `init`, `create-store`, `list-stores`, `use`
- [x] meta.db stores table (MetaStore)
- [x] SigChain → validate entry.store_id
---
## Milestone 1.5: DAG Conflict Resolution
**Goal:** Upgrade store from simple LWW to DAG-based conflict resolution per architecture.md.
### Deliverables
- [x] Proto: Add `repeated bytes parent_hashes` to Entry (for DAG causality)
- [x] Proto: Add `HeadInfo` message for multi-head storage
- [x] Store: KV table schema → `Vec<u8> → Vec<HeadInfo>`
- [x] Store: `apply_entry` → track multiple heads, merge parent tips
- [x] Store: `get` → deterministic winner (highest HLC, author tiebreaker)
- [x] Store: `get_heads` → inspect all heads for a key
- [x] EntryBuilder: `.parent_hashes(...)` method for DAG ancestry
- [x] CLI: Show conflict indicator when multiple heads
### Success Criteria
- [x] Concurrent writes to same key create multiple heads
- [x] Reads return deterministic winner
- [x] Next write citing both heads merges fork to single tip
- [x] All existing tests still pass (71 tests)
---
## Milestone 1.9: Async Refactor
**Goal:** Prepare codebase for concurrent CLI + network operation.
### Deliverables
**Phase 1: Store Actor (sync)**
- [x] Store actor pattern: dedicated thread owns Store, receives commands via `std::sync::mpsc`
- [x] StoreHandle wraps channel sender, keeps current API
- [x] Validate: CLI works as before with actor
**Phase 2: Async Runtime**
- [x] Add tokio runtime (`#[tokio::main]`)
- [x] Migrate `std::sync::mpsc``tokio::sync::mpsc`
- [x] Async CLI using `block_in_place` for sync handlers
### Success Criteria
- [x] CLI still works as before
- [x] Store operations serialized (no data races)
- [x] Ready for concurrent network tasks
---
@@ -39,38 +87,131 @@ Current code assumes single store. Changes needed:
### Deliverables
- [ ] Store: add `applied_frontiers` table (sync state per author)
- [ ] VectorClock module (diff, merge, missing entries)
- [ ] Sync protocol (push missing entries)
- [ ] Iroh integration (peer discovery, connection)
- [ ] Multi-author log merging
- [ ] CLI: `peers`, `connect`/`join` commands
**Phase 1: Sync Logic (no network)**
- [x] SyncState with AuthorInfo (seq + hash) for hash-based log resumption
- [x] `Store::sync_state()` → author-to-seq+hash map from AUTHOR_TABLE
- [x] `SyncState::diff()``Vec<MissingRange>` with from_hash for `read_entries_after`
- [x] Multi-store sync test: compute diff, fetch entries, apply, verify same state
**Phase 2: Iroh Integration**
*Completed:*
- [x] Node info in root store on init: `/nodes/{pubkey}/info` + `/status`
- [x] CLI: `invite <pubkey>` to authorize peers
- [x] CLI: `peers` to list known nodes (with name/added_at info, sorted)
- [x] CLI: `remove <pubkey>` to remove a peer
- [x] Iroh endpoint on startup (same Ed25519 key, mDNS + DNS discovery)
- [x] CLI: `join <nodeid>` - connects to peer, verifies invited
- [x] Peer verification via `/nodes/{pubkey}/status` check
*Join Protocol (new→existing):*
- [x] Proto: `JoinRequest` / `JoinResponse` with store UUID
- [x] Accept handler sends root store UUID in response
- [x] Join command creates empty store with received UUID (no writes until sync)
*Sync Protocol (bidirectional):*
- [x] Proto: `PeerMessage` wrapper with `oneof` for message type discrimination
- [x] `framing.rs` with `MessageSink`/`MessageStream` using `LengthDelimitedCodec`
- [x] Proto: `SyncRequest`/`SyncResponse` using `SyncState`
- [x] `Store::read_entries_after(hash)` to fetch log chunks
- [x] Accept handler: receive SyncState, compute diff, send missing entries
- [x] Sync command: receive entries, apply to store via `apply_entry`
- [x] CLI: `sync [nodeid]` command (syncs with all active peers if no nodeid)
- [x] After sync: node updates own `/nodes/{pubkey}/info` with hostname
*Cleanup*:
- [x] Move core logic from cmd_join and cmd_sync out of commands.rs (now in `sync.rs`)
- [x] Add 'invited' state: invite sets 'invited', peer sets 'active' after sync
*Regressions:*
- [x] Entry ordering: Per-author streaming is correct (hash chain per author, HLC for cross-author).
- [x] Multi-head sync fixed: SyncState now tracks HashSet of head hashes per author.
- [x] Sync entry ordering: Entries sent in HLC order (merge-sort across authors) to ensure causal order.
- [x] `join_mesh` doesn't populate `node.root_store`: Fixed with `complete_join` method.
### Success Criteria
- Node A writes, Node B syncs, both have same state
- Works offline-first (sync when connected)
**Post-M2 Refactoring:**
- [x] Unify `node.rs` from `lattice-cli` and `lattice-core`
- [x] Move network code to `lattice-net`
---
## Milestone 3: Multi-Node Mesh
**Goal:** N nodes form a gossip mesh with watermark consensus.
**Goal:** N nodes form a gossip mesh for real-time sync.
### Deliverables
- [ ] Gossip protocol
- [ ] Watermark tracking & log pruning
- [ ] Node invitation (sigchain membership)
- [ ] Conflict detection (LWW resolution)
**Phase 1: LatticeServer Refactor**
- [x] `LatticeServer` struct in `lattice-net` wrapping `Arc<Node>` + `Endpoint`
- [x] Move `join_mesh`, `sync_with_peer`, `sync_all` to `LatticeServer` methods
- [x] Encapsulate accept loop inside `LatticeServer` (via Router + ProtocolHandler)
- [x] CLI uses `LatticeServer` instead of raw `Node` + `Endpoint`
- [ ] Integration test: invite → join → sync end-to-end
- [ ] Periodic background sync with known peers
- [ ] Track last sync time per peer
**Phase 2: Gossip Protocol** ✓ (iroh-gossip)
- [x] Router handles both `lattice-sync/1` and `/iroh-gossip/1` ALPNs
- [x] `NodeEvent::RootStoreActivated` emitted when root store opens
- [x] Auto-join gossip topic on root store activation
- [x] Broadcast local entries to gossip topic on commit
- [x] Receive gossip entries and apply to store
- [x] Topic ID via `blake3::hash("lattice/{store_id}")`
- [ ] Gossip bootstrap peers from `/peers/` (needs Prefix Watch)
**Next: Prefix Watch (reactive store updates)**
- [ ] `store.watch_prefix(prefix) -> Receiver<WatchEvent>`
- [ ] `WatchEvent::Put { key, value }` / `WatchEvent::Delete { key }`
- [ ] StoreActor tracks watchers per prefix, emits on matching put/delete
- [ ] LatticeServer uses `/peers/` watch to update gossip bootstrap peers dynamically
- [ ] Enables reactive patterns: config changes, presence, app-level subscriptions
---
## Technical Debt
**Logging**
- [ ] Replace `println!`/`eprintln!` with `tracing` crate (`tracing::info!`, `tracing::error!`)
- Standard in Rust async ecosystem, used by Iroh internally
**Lifecycle Management (Zombie Tasks)**
- [ ] Spawned infinite loops (`spawn_node_event_listener`, `spawn_entry_forward_loop`, gossip receive loop) keep running if `LatticeServer` is dropped
- [ ] Use `tokio_util::sync::CancellationToken` or keep `JoinHandle`s for graceful shutdown
**Error Handling**
- [ ] Replace `Result<..., String>` with `anyhow::Result` or define `LatticeNetError` enum
- String errors make it hard to handle specific failure cases
---
## Future
- offline nodes should not delay sync
- sync command should transitive sync all peers
- Gossip:
- gossip new entries to peers
- backfill missing entries from peers (how do peers notice missing entries?)
- snapshots for kv store
- prune using consensus watermark
- remove_peer should be a transactional operation on store
- Watermark tracking & log pruning
- Track minimum confirmed seq per author across all peers
- Log pruning: remove entries below watermark
- Multi-KV-Store sync
- Optimized sync on join. Only transfer current watermark state, then sync missing entries. This would allow pruning. Might need snapshot support in KV store.
- Mobile (iOS/Android) clients
- Key rotation
- Secure storage (Keychain, TPM)
- Snapshots for fast bootstrap
- FUSE filesystem mount
- Note: FUSE requires u64 inode numbers → maintain `BiMap<u64, Hash>` in redb
- Merkle-ized State
- state.db as Merkle tree with signed root hash
- O(1) sync checks (compare root), efficient binary-search diffing
- Light clients: fetch value + Merkle proof, verify without full state
- Trade-off: write amplification, requires deterministic tree (Patricia Trie / Merkle Search Tree)
+39
View File
@@ -0,0 +1,39 @@
# Storage Format
## Directory Layout
```
~/.local/share/lattice/
├── identity.key # Ed25519 private key (not replicated)
├── meta.db # Global metadata (redb)
└── stores/{uuid}/
├── logs/{author}.log # Append-only SignedEntry stream
└── state.db # Per-store KV state (redb)
```
## meta.db (redb)
| Table | Key | Value | Purpose |
|---------|---------------|--------------------|------------------------------|
| stores | UUID (16B) | created_at (u64) | Known stores |
| meta | "root_store" | UUID (16B) | Auto-opened on CLI startup |
## state.db (redb, per store)
| Table | Key | Value | Purpose |
|---------|----------|-------------|------------------------|
| kv | String | Vec<u8> | Key-value data |
| meta | String | Vec<u8> | last_seq, last_hash |
## Log Files
Each `{author}.log` contains length-delimited `LogRecord` messages:
```protobuf
message LogRecord {
bytes hash = 1; // BLAKE3 hash of entry_bytes
bytes entry_bytes = 2; // Serialized SignedEntry
}
```
Hashes are verified on read; corruption causes `LogError::HashMismatch`.
+25
View File
@@ -0,0 +1,25 @@
[package]
name = "lattice-cli"
description = "Interactive CLI for Lattice"
version.workspace = true
edition.workspace = true
license.workspace = true
[[bin]]
name = "lattice"
path = "src/main.rs"
[dependencies]
lattice-core = { workspace = true }
lattice-net = { workspace = true }
rustyline = { workspace = true }
hex = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true }
shlex = "1"
serde_json = "1"
iroh = { workspace = true }
prost = { workspace = true }
chrono = { workspace = true }
tokio-util = { version = "0.7", features = ["codec"] }
futures-util = "0.3"
+79
View File
@@ -0,0 +1,79 @@
//! CLI command handlers
use lattice_core::{Node, StoreHandle};
use lattice_net::LatticeServer;
/// Result of a command that may switch stores or exit
pub enum CommandResult {
/// No store change
Ok,
/// Switch to this store
SwitchTo(StoreHandle),
/// Exit the CLI
Quit,
}
/// Helper to call async code from sync command handlers
pub fn block_async<F: std::future::Future>(f: F) -> F::Output {
tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(f))
}
pub type Handler = fn(&Node, Option<&StoreHandle>, Option<&LatticeServer>, &[String]) -> CommandResult;
pub struct Command {
pub name: &'static str,
pub args: &'static str,
pub desc: &'static str,
pub group: &'static str,
pub min_args: usize,
pub max_args: usize,
pub handler: Handler,
}
/// Get all available commands
pub fn commands() -> Vec<Command> {
let mut cmds = Vec::new();
// General CLI commands
cmds.push(Command {
name: "help", args: "", desc: "Show this help",
group: "general", min_args: 0, max_args: 0, handler: cmd_help as Handler
});
cmds.push(Command {
name: "quit", args: "", desc: "Exit",
group: "general", min_args: 0, max_args: 0, handler: cmd_quit as Handler
});
// Node commands (operations on the node)
cmds.extend(crate::node_commands::node_commands());
// Store commands (raw KV operations)
cmds.extend(crate::store_commands::store_commands());
cmds
}
fn cmd_help(_node: &Node, _store: Option<&StoreHandle>, _server: Option<&LatticeServer>, _args: &[String]) -> CommandResult {
let cmds = commands();
let mut last_group = "";
for cmd in &cmds {
if cmd.group != last_group {
println!();
println!("[{}]", cmd.group);
last_group = cmd.group;
}
let usage = if cmd.args.is_empty() {
cmd.name.to_string()
} else {
format!("{} {}", cmd.name, cmd.args)
};
println!(" {:18} {}", usage, cmd.desc);
}
println!();
CommandResult::Ok
}
fn cmd_quit(_node: &Node, _store: Option<&StoreHandle>, _server: Option<&LatticeServer>, _args: &[String]) -> CommandResult {
println!("Goodbye!");
CommandResult::Quit
}
+121
View File
@@ -0,0 +1,121 @@
//! Lattice Interactive CLI
mod commands;
mod node_commands;
mod store_commands;
use lattice_net::LatticeServer;
use commands::CommandResult;
use lattice_core::{NodeBuilder, StoreHandle};
use rustyline::error::ReadlineError;
use rustyline::DefaultEditor;
use std::sync::Arc;
#[tokio::main]
async fn main() {
println!("Lattice CLI v{}", env!("CARGO_PKG_VERSION"));
println!("Type 'help' for commands, 'quit' to exit.\n");
let node = match NodeBuilder::new().build() {
Ok(n) => Arc::new(n),
Err(e) => {
eprintln!("Failed to initialize: {}", e);
return;
}
};
// Create LatticeServer (creates endpoint and spawns accept loop internally)
let server = match LatticeServer::new_from_node(node.clone()).await {
Ok(s) => {
println!("Iroh: {} (listening)", s.endpoint().public_key().fmt_short());
Some(s)
}
Err(e) => {
eprintln!("Warning: Iroh failed to start: {}", e);
None
}
};
let info = node.info();
println!("Node ID: {}", info.node_id);
println!("Data: {}", info.data_path);
if !info.stores.is_empty() {
println!("Stores: {}", info.stores.len());
}
let mut current_store: Option<StoreHandle> = match node.open_root_store().await {
Ok(Some(open_info)) => {
if open_info.entries_replayed > 0 {
println!("Root: {} (replayed {})", open_info.store_id, open_info.entries_replayed);
} else {
println!("Root: {}", open_info.store_id);
}
node.root_store().await.as_ref().cloned()
}
Ok(None) => {
println!("Status: Not initialized (use 'init')");
None
}
Err(e) => {
eprintln!("Warning: {}", e);
None
}
};
println!();
let mut rl = DefaultEditor::new().expect("Failed to create editor");
let cmds = commands::commands();
loop {
let prompt = match &current_store {
Some(h) => format!("lattice:{}> ", &h.id().to_string()[..8]),
None => "lattice:no-store> ".to_string(),
};
match rl.readline(&prompt) {
Ok(line) => {
let line = line.trim();
if line.is_empty() { continue; }
let _ = rl.add_history_entry(line);
let args = match shlex::split(line) {
Some(a) => a,
None => {
println!("Error: mismatched quotes");
continue;
}
};
let cmd_name = args.first().map(|s| s.as_str()).unwrap_or("");
match cmds.iter().find(|c| c.name == cmd_name || (cmd_name == "exit" && c.name == "quit")) {
Some(cmd) => {
let cmd_args = &args[1..];
if cmd_args.len() < cmd.min_args || cmd_args.len() > cmd.max_args {
println!("Usage: {} {}", cmd.name, cmd.args);
} else {
match (cmd.handler)(&node, current_store.as_ref(), server.as_ref(), cmd_args) {
CommandResult::Ok => {}
CommandResult::SwitchTo(h) => {
current_store = Some(h);
}
CommandResult::Quit => break,
}
}
}
None => println!("Unknown: '{}'. Type 'help'.", cmd_name),
}
}
Err(ReadlineError::Interrupted | ReadlineError::Eof) => {
println!("Goodbye!");
break;
}
Err(e) => {
eprintln!("Error: {:?}", e);
break;
}
}
}
}
+313
View File
@@ -0,0 +1,313 @@
//! Node commands - operations on the node (mesh, peers, status)
use crate::commands::{block_async, Command, CommandResult, Handler};
use lattice_core::{Node, StoreHandle, PeerStatus, Uuid};
use lattice_net::LatticeServer;
use chrono::DateTime;
use std::time::Instant;
pub fn node_commands() -> Vec<Command> {
vec![
// Store management
Command { name: "init", args: "", desc: "Initialize root store", group: "node", min_args: 0, max_args: 0, handler: cmd_init as Handler },
Command { name: "create-store", args: "", desc: "Create a new store", group: "node", min_args: 0, max_args: 0, handler: cmd_create_store as Handler },
Command { name: "use", args: "<uuid>", desc: "Switch to a store", group: "node", min_args: 1, max_args: 1, handler: cmd_use_store as Handler },
Command { name: "list-stores", args: "", desc: "List all stores", group: "node", min_args: 0, max_args: 0, handler: cmd_list_stores as Handler },
Command { name: "node-status", args: "", desc: "Show node info", group: "node", min_args: 0, max_args: 0, handler: cmd_node_status as Handler },
// Peer management
Command { name: "invite", args: "<pubkey>", desc: "Invite a peer", group: "peers", min_args: 1, max_args: 1, handler: cmd_invite as Handler },
Command { name: "peers", args: "", desc: "List all peers", group: "peers", min_args: 0, max_args: 0, handler: cmd_peers as Handler },
Command { name: "remove", args: "<pubkey>", desc: "Remove a peer", group: "peers", min_args: 1, max_args: 1, handler: cmd_remove as Handler },
// Networking
Command { name: "join", args: "<node_id>", desc: "Join an existing mesh", group: "network", min_args: 1, max_args: 1, handler: cmd_join as Handler },
Command { name: "sync", args: "[node_id]", desc: "Sync with peers", group: "network", min_args: 0, max_args: 1, handler: cmd_sync as Handler },
]
}
// --- Store management ---
fn cmd_init(node: &Node, _store: Option<&StoreHandle>, _server: Option<&LatticeServer>, _args: &[String]) -> CommandResult {
match block_async(node.init()) {
Ok(store_id) => {
println!("Initialized with root store: {}", store_id);
println!("Node info stored in /nodes/{}/*", hex::encode(node.node_id()));
match block_async(node.root_store()).as_ref() {
Some(h) => CommandResult::SwitchTo(h.clone()),
None => CommandResult::Ok,
}
}
Err(e) => {
eprintln!("Error: {}", e);
CommandResult::Ok
}
}
}
fn cmd_create_store(node: &Node, _store: Option<&StoreHandle>, _server: Option<&LatticeServer>, _args: &[String]) -> CommandResult {
match node.create_store() {
Ok(store_id) => {
println!("Created store: {}", store_id);
match block_async(node.open_store(store_id)) {
Ok((handle, _)) => {
println!("Switched to new store");
CommandResult::SwitchTo(handle)
}
Err(e) => {
eprintln!("Warning: {}", e);
CommandResult::Ok
}
}
}
Err(e) => {
eprintln!("Error: {}", e);
CommandResult::Ok
}
}
}
fn cmd_use_store(node: &Node, _store: Option<&StoreHandle>, _server: Option<&LatticeServer>, args: &[String]) -> CommandResult {
let store_id = match Uuid::parse_str(&args[0]) {
Ok(id) => id,
Err(_) => {
eprintln!("Error: invalid UUID '{}'", args[0]);
return CommandResult::Ok;
}
};
let start = Instant::now();
match block_async(node.open_store(store_id)) {
Ok((handle, info)) => {
if info.entries_replayed > 0 {
println!("Replayed {} entries ({:.2?})", info.entries_replayed, start.elapsed());
} else {
println!("Switched to store {}", store_id);
}
CommandResult::SwitchTo(handle)
}
Err(e) => {
eprintln!("Error: {}", e);
CommandResult::Ok
}
}
}
fn cmd_list_stores(node: &Node, store: Option<&StoreHandle>, _server: Option<&LatticeServer>, _args: &[String]) -> CommandResult {
let stores = match node.list_stores() {
Ok(s) => s,
Err(e) => {
eprintln!("Error: {}", e);
return CommandResult::Ok;
}
};
let current_id = store.map(|s| s.id());
if stores.is_empty() {
println!("No stores. Use 'init' or 'create-store'.");
} else {
for store_id in stores {
let marker = if Some(store_id) == current_id { " *" } else { "" };
println!("{}{}", store_id, marker);
}
}
CommandResult::Ok
}
// --- Info ---
fn cmd_node_status(node: &Node, _store: Option<&StoreHandle>, _server: Option<&LatticeServer>, _args: &[String]) -> CommandResult {
println!("Node ID: {}", hex::encode(node.node_id()));
if let Some(name) = node.name() {
println!("Name: {}", name);
}
println!("Data: {}", node.data_path().display());
match node.root_store_id() {
Ok(Some(id)) => println!("Root: {}", id),
Ok(None) => println!("Root: (not set)"),
Err(_) => println!("Root: (error)"),
}
// Count peers using node.list_peers()
if let Ok(peers) = block_async(node.list_peers()) {
let active = peers.iter().filter(|p| p.status == PeerStatus::Active).count();
let invited = peers.iter().filter(|p| p.status == PeerStatus::Invited).count();
println!("Peers: {} active, {} invited", active, invited);
}
CommandResult::Ok
}
// --- Peer management ---
fn cmd_invite(node: &Node, _store: Option<&StoreHandle>, _server: Option<&LatticeServer>, args: &[String]) -> CommandResult {
let pubkey_hex = &args[0];
let pubkey: [u8; 32] = match hex::decode(pubkey_hex) {
Ok(bytes) if bytes.len() == 32 => bytes.try_into().unwrap(),
_ => {
eprintln!("Invalid pubkey: expected 64 hex chars (32 bytes)");
return CommandResult::Ok;
}
};
match block_async(node.invite_peer(&pubkey)) {
Ok(()) => {
println!("Invited peer: {}", pubkey_hex);
println!(" Status: {} (will become active after sync)", PeerStatus::Invited.as_str());
}
Err(e) => eprintln!("Error: {}", e),
}
CommandResult::Ok
}
fn cmd_peers(node: &Node, _store: Option<&StoreHandle>, _server: Option<&LatticeServer>, _args: &[String]) -> CommandResult {
let peers = match block_async(node.list_peers()) {
Ok(p) => p,
Err(e) => {
eprintln!("Error: {}", e);
return CommandResult::Ok;
}
};
if peers.is_empty() {
println!("No peers found.");
return CommandResult::Ok;
}
// Group peers by status
let mut by_status: std::collections::HashMap<PeerStatus, Vec<&lattice_core::PeerInfo>> =
std::collections::HashMap::new();
for peer in &peers {
by_status.entry(peer.status).or_default().push(peer);
}
// Print grouped by status in order: active, invited, dormant
let status_order = [PeerStatus::Active, PeerStatus::Invited, PeerStatus::Dormant];
for status in &status_order {
if let Some(peer_list) = by_status.get(status) {
println!("\n[{}] ({}):", status.as_str(), peer_list.len());
let mut sorted: Vec<_> = peer_list.iter().collect();
sorted.sort_by(|a, b| a.pubkey.cmp(&b.pubkey));
for peer in sorted {
let added_str = peer.added_at
.and_then(|ts| DateTime::from_timestamp(ts as i64, 0))
.map(|dt| dt.format("%Y-%m-%d").to_string())
.unwrap_or_default();
let info_str = match (peer.name.as_ref(), added_str.is_empty()) {
(Some(name), false) => format!(" {} ({})", name, added_str),
(Some(name), true) => format!(" {}", name),
(None, false) => format!(" ({})", added_str),
(None, true) => String::new(),
};
println!(" {}{}", peer.pubkey, info_str);
}
}
}
CommandResult::Ok
}
fn cmd_remove(node: &Node, _store: Option<&StoreHandle>, _server: Option<&LatticeServer>, args: &[String]) -> CommandResult {
let pubkey_hex = &args[0];
let pubkey: [u8; 32] = match hex::decode(pubkey_hex) {
Ok(bytes) if bytes.len() == 32 => bytes.try_into().unwrap(),
_ => {
eprintln!("Invalid pubkey: expected 64 hex characters");
return CommandResult::Ok;
}
};
match block_async(node.remove_peer(&pubkey)) {
Ok(()) => println!("Removed peer: {}...", &pubkey_hex[..10]),
Err(e) => eprintln!("Error: {}", e),
}
CommandResult::Ok
}
// --- Networking ---
fn cmd_join(_node: &Node, store: Option<&StoreHandle>, server: Option<&LatticeServer>, args: &[String]) -> CommandResult {
let server = match server {
Some(s) => s,
None => {
eprintln!("Iroh endpoint not started.");
return CommandResult::Ok;
}
};
if store.is_some() {
eprintln!("Already initialized. Use 'sync' to sync with peers.");
return CommandResult::Ok;
}
let peer_id = match lattice_net::parse_node_id(&args[0]) {
Ok(id) => id,
Err(e) => {
eprintln!("Invalid node ID: {}", e);
return CommandResult::Ok;
}
};
println!("Joining mesh via {}...", peer_id.fmt_short());
match block_async(server.join_mesh(peer_id)) {
Ok(handle) => {
println!("Joined mesh! Use 'sync' command to sync entries.");
CommandResult::SwitchTo(handle)
}
Err(e) => {
eprintln!("Join failed: {}", e);
CommandResult::Ok
}
}
}
fn cmd_sync(_node: &Node, store: Option<&StoreHandle>, server: Option<&LatticeServer>, args: &[String]) -> CommandResult {
let server = match server {
Some(s) => s,
None => {
eprintln!("Iroh endpoint not started.");
return CommandResult::Ok;
}
};
let store = match store {
Some(s) => s,
None => {
eprintln!("No store open. Use 'init' or 'join' first.");
return CommandResult::Ok;
}
};
if args.is_empty() {
// Sync with all active peers
match block_async(server.sync_all(store)) {
Ok(results) => {
if results.is_empty() {
println!("No peers to sync with.");
} else {
let total: u64 = results.iter().map(|r| r.entries_applied).sum();
println!("\nSync complete! Applied {} entries from {} peer(s).", total, results.len());
}
}
Err(e) => eprintln!("Sync failed: {}", e),
}
} else {
// Sync with specific peer
let peer_id = match lattice_net::parse_node_id(&args[0]) {
Ok(id) => id,
Err(e) => {
eprintln!("Invalid node ID: {}", e);
return CommandResult::Ok;
}
};
println!("Syncing with {}...", peer_id.fmt_short());
match block_async(server.sync_with_peer(store, peer_id)) {
Ok(result) => {
println!("Sync complete! Applied {} entries (peer sent {})",
result.entries_applied, result.entries_sent_by_peer);
}
Err(e) => eprintln!("Sync failed: {}", e),
}
}
CommandResult::Ok
}
+220
View File
@@ -0,0 +1,220 @@
//! Store commands - direct KV operations
use crate::commands::{block_async, Command, CommandResult, Handler};
use lattice_core::{Node, StoreHandle};
use lattice_net::LatticeServer;
use std::time::Instant;
pub fn store_commands() -> Vec<Command> {
vec![
Command { name: "store-status", args: "", desc: "Show store info", group: "store", min_args: 0, max_args: 0, handler: cmd_store_status as Handler },
Command { name: "put", args: "<key> <value>", desc: "Store a key-value pair", group: "store", min_args: 2, max_args: 2, handler: cmd_put as Handler },
Command { name: "get", args: "<key> [-v]", desc: "Get value for key", group: "store", min_args: 1, max_args: 2, handler: cmd_get as Handler },
Command { name: "delete", args: "<key>", desc: "Delete a key", group: "store", min_args: 1, max_args: 1, handler: cmd_delete as Handler },
Command { name: "list", args: "[prefix] [-v]", desc: "List keys (optionally filtered by prefix)", group: "store", min_args: 0, max_args: 2, handler: cmd_list as Handler },
Command { name: "author-state", args: "[pubkey]", desc: "Show author sync state", group: "store", min_args: 0, max_args: 1, handler: cmd_author_state as Handler },
]
}
fn cmd_store_status(_node: &Node, store: Option<&StoreHandle>, _server: Option<&LatticeServer>, _args: &[String]) -> CommandResult {
let Some(h) = store else {
println!("No store selected. Use 'init' or 'use <uuid>'");
return CommandResult::Ok;
};
println!("Store ID: {}", h.id());
println!("Log Seq: {}", block_async(h.log_seq()));
println!("Applied: {}", block_async(h.applied_seq()).unwrap_or(0));
let all = block_async(h.list(false)).unwrap_or_default();
println!("Keys: {}", all.len());
// Show log directory size
let (file_count, total_size) = block_async(h.log_stats());
if file_count > 0 {
println!("Logs: {} files, {} bytes", file_count, total_size);
}
CommandResult::Ok
}
fn cmd_put(_node: &Node, store: Option<&StoreHandle>, _server: Option<&LatticeServer>, args: &[String]) -> CommandResult {
let Some(h) = store else {
println!("No store selected. Use 'init' or 'use <uuid>'");
return CommandResult::Ok;
};
let start = Instant::now();
match block_async(h.put(args[0].as_bytes(), args[1].as_bytes())) {
Ok(seq) => println!("OK (seq: {}, {:.2?})", seq, start.elapsed()),
Err(e) => eprintln!("Error: {}", e),
}
CommandResult::Ok
}
fn cmd_get(_node: &Node, store: Option<&StoreHandle>, _server: Option<&LatticeServer>, args: &[String]) -> CommandResult {
let Some(h) = store else {
println!("No store selected. Use 'init' or 'use <uuid>'");
return CommandResult::Ok;
};
let verbose = args.get(1).map(|a| a == "-v").unwrap_or(false);
let start = Instant::now();
let key = args[0].as_bytes();
if verbose {
// Show all heads
match block_async(h.get_heads(key)) {
Ok(heads) if heads.is_empty() => println!("(nil)"),
Ok(heads) => {
for (i, head) in heads.iter().enumerate() {
let winner = if i == 0 { "" } else { " " };
let tombstone = if head.tombstone { "" } else { "" };
let author_short = hex::encode(&head.author).chars().take(8).collect::<String>();
if head.tombstone {
println!("{} {} (deleted) (hlc:{}, author:{})",
winner, tombstone, head.hlc, author_short);
} else {
println!("{} {} (hlc:{}, author:{})",
winner, format_value(&head.value), head.hlc, author_short);
}
}
if heads.len() > 1 {
println!("{} heads (conflict)", heads.len());
}
println!("({:.2?})", start.elapsed());
}
Err(e) => eprintln!("Error: {}", e),
}
} else {
match block_async(h.get(key)) {
Ok(Some(v)) => {
let heads = block_async(h.get_heads(key)).unwrap_or_default();
if heads.len() > 1 {
println!("{} (⚠ {} heads)", format_value(&v), heads.len());
} else {
println!("{}", format_value(&v));
}
println!("({:.2?})", start.elapsed());
}
Ok(None) => println!("(nil)"),
Err(e) => eprintln!("Error: {}", e),
}
}
CommandResult::Ok
}
fn cmd_delete(_node: &Node, store: Option<&StoreHandle>, _server: Option<&LatticeServer>, args: &[String]) -> CommandResult {
let Some(h) = store else {
println!("No store selected. Use 'init' or 'use <uuid>'");
return CommandResult::Ok;
};
let start = Instant::now();
match block_async(h.delete(args[0].as_bytes())) {
Ok(seq) => println!("OK (seq: {}, {:.2?})", seq, start.elapsed()),
Err(e) => eprintln!("Error: {}", e),
}
CommandResult::Ok
}
fn cmd_list(_node: &Node, store: Option<&StoreHandle>, _server: Option<&LatticeServer>, args: &[String]) -> CommandResult {
let Some(h) = store else {
println!("No store selected. Use 'init' or 'use <uuid>'");
return CommandResult::Ok;
};
// Parse args: [prefix] [-v]
let verbose = args.iter().any(|a| a == "-v");
let prefix = args.iter().find(|a| *a != "-v").cloned();
let start = Instant::now();
let result = if let Some(p) = &prefix {
block_async(h.list_by_prefix(p.as_bytes(), verbose))
} else {
block_async(h.list(verbose))
};
match result {
Ok(entries) => {
if entries.is_empty() {
println!("(empty)");
} else {
for (k, v) in &entries {
let key_str = format_value(k);
if verbose {
// Show all heads for this key
let heads = block_async(h.get_heads(k)).unwrap_or_default();
println!("{}:", key_str);
for (i, head) in heads.iter().enumerate() {
let winner = if i == 0 { "" } else { " " };
let author_short = hex::encode(&head.author).chars().take(8).collect::<String>();
if head.tombstone {
println!(" {} ⊗ (deleted) (hlc:{}, author:{})",
winner, head.hlc, author_short);
} else {
println!(" {} {} (hlc:{}, author:{})",
winner, format_value(&head.value), head.hlc, author_short);
}
}
} else {
// Check for multiple heads
let heads = block_async(h.get_heads(k)).unwrap_or_default();
if heads.len() > 1 {
println!("{} = {} (⚠ {} heads)", key_str, format_value(v), heads.len());
} else {
println!("{} = {}", key_str, format_value(v));
}
}
}
let prefix_str = prefix.as_ref().map(|p| format!(" (prefix: {})", p)).unwrap_or_default();
println!("({} keys{}, {:.2?})", entries.len(), prefix_str, start.elapsed());
}
}
Err(e) => eprintln!("Error: {}", e),
}
CommandResult::Ok
}
fn cmd_author_state(node: &Node, store: Option<&StoreHandle>, _server: Option<&LatticeServer>, args: &[String]) -> CommandResult {
let store = match store {
Some(s) => s,
None => {
eprintln!("Error: no store selected");
return CommandResult::Ok;
}
};
// Get author: from arg or default to self
let author_bytes: [u8; 32] = if args.is_empty() {
node.node_id()
} else {
let hex_str = args[0].trim_start_matches("0x");
match hex::decode(hex_str) {
Ok(bytes) if bytes.len() == 32 => bytes.try_into().unwrap(),
Ok(bytes) => {
eprintln!("Error: author must be 32 bytes, got {}", bytes.len());
return CommandResult::Ok;
}
Err(e) => {
eprintln!("Error: invalid hex: {}", e);
return CommandResult::Ok;
}
}
};
match block_async(store.author_state(&author_bytes)) {
Ok(Some(state)) => {
println!("Author: {}", hex::encode(&author_bytes));
println!(" seq: {}", state.seq);
println!(" hash: {}", hex::encode(&state.hash));
println!(" log_offset: {}", state.log_offset);
}
Ok(None) => {
println!("No state for author: {}", hex::encode(&author_bytes));
}
Err(e) => eprintln!("Error: {}", e),
}
CommandResult::Ok
}
fn format_value(v: &[u8]) -> String {
std::str::from_utf8(v).map(String::from).unwrap_or_else(|_| format!("0x{}", hex::encode(v)))
}
+5
View File
@@ -14,6 +14,11 @@ bytes = { workspace = true }
dirs = { workspace = true }
blake3 = { workspace = true }
hex = { workspace = true }
redb = { workspace = true }
uuid = { workspace = true }
tokio = { workspace = true }
hostname = "0.4"
serde_json = "1"
[build-dependencies]
prost-build = { workspace = true }
+189
View File
@@ -0,0 +1,189 @@
//! Causal Entry Iterator - yields entries in HLC (causal) order
//!
//! Implements merge-sort streaming across multiple author queues using a min-heap,
//! ensuring entries are returned in correct causal order for sync.
//! Complexity: O(N log K) where N = total entries, K = number of authors.
use crate::proto::{Entry, SignedEntry};
use prost::Message;
use std::cmp::Ordering;
use std::collections::{BinaryHeap, VecDeque};
/// A heap entry that wraps an author queue index and the HLC of its front entry.
/// Uses Reverse for min-heap behavior (lowest HLC first).
struct HeapEntry {
hlc: (u64, u32),
queue_idx: usize,
}
impl PartialEq for HeapEntry {
fn eq(&self, other: &Self) -> bool {
self.hlc == other.hlc
}
}
impl Eq for HeapEntry {}
impl PartialOrd for HeapEntry {
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
Some(self.cmp(other))
}
}
impl Ord for HeapEntry {
fn cmp(&self, other: &Self) -> Ordering {
// Reverse order for min-heap (BinaryHeap is max-heap by default)
other.hlc.cmp(&self.hlc)
}
}
/// Iterator that yields SignedEntry in HLC (causal) order.
///
/// Takes multiple VecDeques (one per author) and yields entries
/// from lowest to highest HLC, ensuring causal ordering for sync.
/// Uses a min-heap for O(log K) per-entry overhead instead of O(K) linear scan.
pub struct CausalEntryIter {
queues: Vec<VecDeque<SignedEntry>>,
heap: BinaryHeap<HeapEntry>,
}
impl CausalEntryIter {
/// Create a new iterator from a list of entry queues (one per author)
pub fn new(queues: Vec<VecDeque<SignedEntry>>) -> Self {
let mut heap = BinaryHeap::with_capacity(queues.len());
// Initialize heap with the front entry from each non-empty queue
for (idx, queue) in queues.iter().enumerate() {
if let Some(entry) = queue.front() {
heap.push(HeapEntry {
hlc: Self::get_hlc(entry),
queue_idx: idx,
});
}
}
Self { queues, heap }
}
/// Extract HLC (wall_time, counter) from a SignedEntry
fn get_hlc(entry: &SignedEntry) -> (u64, u32) {
Entry::decode(&entry.entry_bytes[..])
.ok()
.and_then(|e| e.timestamp)
.map(|t| (t.wall_time, t.counter))
.unwrap_or((0, 0))
}
}
impl Iterator for CausalEntryIter {
type Item = SignedEntry;
fn next(&mut self) -> Option<Self::Item> {
// Pop the queue with lowest HLC
let HeapEntry { queue_idx, .. } = self.heap.pop()?;
// Remove entry from that queue
let entry = self.queues[queue_idx].pop_front()?;
// If queue still has entries, push its new front back to heap
if let Some(next_entry) = self.queues[queue_idx].front() {
self.heap.push(HeapEntry {
hlc: Self::get_hlc(next_entry),
queue_idx,
});
}
Some(entry)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::hlc::HLC;
use crate::clock::MockClock;
use crate::node_identity::NodeIdentity;
use crate::signed_entry::EntryBuilder;
fn make_entry(node: &NodeIdentity, seq: u64, clock_ms: u64) -> SignedEntry {
let clock = MockClock::new(clock_ms);
EntryBuilder::new(seq, HLC::now_with_clock(&clock))
.store_id(vec![0u8; 16])
.prev_hash(vec![0u8; 32])
.put(b"/test".to_vec(), format!("seq{}", seq).into_bytes())
.sign(node)
}
#[test]
fn test_empty_iter() {
let iter = CausalEntryIter::new(vec![]);
assert_eq!(iter.count(), 0);
}
#[test]
fn test_single_queue() {
let node = NodeIdentity::generate();
let entries: VecDeque<_> = vec![
make_entry(&node, 1, 1000),
make_entry(&node, 2, 2000),
].into();
let iter = CausalEntryIter::new(vec![entries]);
let result: Vec<_> = iter.collect();
assert_eq!(result.len(), 2);
}
#[test]
fn test_merge_multiple_queues() {
let node_a = NodeIdentity::generate();
let node_b = NodeIdentity::generate();
// Author A: entries at time 1000, 3000
let queue_a: VecDeque<_> = vec![
make_entry(&node_a, 1, 1000),
make_entry(&node_a, 2, 3000),
].into();
// Author B: entries at time 2000
let queue_b: VecDeque<_> = vec![
make_entry(&node_b, 1, 2000),
].into();
let iter = CausalEntryIter::new(vec![queue_a, queue_b]);
let result: Vec<_> = iter.collect();
// Should be in HLC order: 1000, 2000, 3000
assert_eq!(result.len(), 3);
// Verify order by checking HLC values
let hlcs: Vec<_> = result.iter()
.map(|e| CausalEntryIter::get_hlc(e))
.collect();
assert_eq!(hlcs[0].0, 1000);
assert_eq!(hlcs[1].0, 2000);
assert_eq!(hlcs[2].0, 3000);
}
#[test]
fn test_many_authors() {
// Test with 10 authors to verify heap behavior
let nodes: Vec<_> = (0..10).map(|_| NodeIdentity::generate()).collect();
let queues: Vec<VecDeque<_>> = nodes.iter().enumerate().map(|(i, node)| {
vec![make_entry(node, 1, (i * 100 + 50) as u64)].into()
}).collect();
let iter = CausalEntryIter::new(queues);
let result: Vec<_> = iter.collect();
assert_eq!(result.len(), 10);
// Verify strictly increasing HLC order
let hlcs: Vec<_> = result.iter()
.map(|e| CausalEntryIter::get_hlc(e).0)
.collect();
for window in hlcs.windows(2) {
assert!(window[0] < window[1], "HLCs should be strictly increasing");
}
}
}
+55 -28
View File
@@ -2,19 +2,26 @@
//!
//! Provides platform-specific paths for Lattice data storage:
//! - `identity.key` — Ed25519 private key
//! - `logs/` — Append-only log files per author
//! - `state.db` — KV snapshot and indexes
//! - `meta.db` — Global metadata (stores table)
//! - `stores/{uuid}/logs/{author}.log` — Per-store, per-author logs
//! - `stores/{uuid}/state.db` — Per-store KV state
use std::path::{Path, PathBuf};
use uuid::Uuid;
const APP_NAME: &str = "lattice";
/// Data directory configuration.
///
/// Handles paths for:
/// - `identity.key` — node's private key
/// - `logs/{author_id}.log` — per-author log files
/// - `state.db` — redb database
/// Multi-store layout:
/// ```text
/// base/
/// identity.key
/// meta.db
/// stores/{uuid}/
/// logs/{author}.log
/// state.db
/// ```
#[derive(Debug, Clone)]
pub struct DataDir {
base: PathBuf,
@@ -27,10 +34,6 @@ impl DataDir {
}
/// Create a DataDir using the platform-specific data directory.
///
/// - Linux: `~/.local/share/lattice/`
/// - macOS: `~/Library/Application Support/lattice/`
/// - Windows: `C:\Users\<user>\AppData\Roaming\lattice\`
pub fn default_location() -> Option<Self> {
dirs::data_dir().map(|d| Self::new(d.join(APP_NAME)))
}
@@ -45,25 +48,47 @@ impl DataDir {
self.base.join("identity.key")
}
/// Get the path to the logs directory.
pub fn logs_dir(&self) -> PathBuf {
self.base.join("logs")
/// Get the path to the global metadata database.
pub fn meta_db(&self) -> PathBuf {
self.base.join("meta.db")
}
/// Get the path to a specific author's log file.
pub fn log_file(&self, author_id_hex: &str) -> PathBuf {
self.logs_dir().join(format!("{}.log", author_id_hex))
/// Get the path to the stores directory.
pub fn stores_dir(&self) -> PathBuf {
self.base.join("stores")
}
/// Get the path to the state database.
pub fn state_db(&self) -> PathBuf {
self.base.join("state.db")
/// Get the path to a specific store's directory.
pub fn store_dir(&self, store_id: Uuid) -> PathBuf {
self.stores_dir().join(store_id.to_string())
}
/// Ensure all required directories exist.
/// Get the path to a store's logs directory.
pub fn store_logs_dir(&self, store_id: Uuid) -> PathBuf {
self.store_dir(store_id).join("logs")
}
/// Get the path to a specific author's log file within a store.
pub fn store_log_file(&self, store_id: Uuid, author_id_hex: &str) -> PathBuf {
self.store_logs_dir(store_id).join(format!("{}.log", author_id_hex))
}
/// Get the path to a store's state database.
pub fn store_state_db(&self, store_id: Uuid) -> PathBuf {
self.store_dir(store_id).join("state.db")
}
/// Ensure base directory exists.
pub fn ensure_dirs(&self) -> std::io::Result<()> {
std::fs::create_dir_all(&self.base)?;
std::fs::create_dir_all(self.logs_dir())?;
std::fs::create_dir_all(self.stores_dir())?;
Ok(())
}
/// Ensure directories for a specific store exist.
pub fn ensure_store_dirs(&self, store_id: Uuid) -> std::io::Result<()> {
self.ensure_dirs()?;
std::fs::create_dir_all(self.store_logs_dir(store_id))?;
Ok(())
}
}
@@ -83,29 +108,31 @@ mod tests {
let dd = DataDir::new("/custom/path");
assert_eq!(dd.base(), Path::new("/custom/path"));
assert_eq!(dd.identity_key(), PathBuf::from("/custom/path/identity.key"));
assert_eq!(dd.logs_dir(), PathBuf::from("/custom/path/logs"));
assert_eq!(dd.state_db(), PathBuf::from("/custom/path/state.db"));
assert_eq!(dd.meta_db(), PathBuf::from("/custom/path/meta.db"));
assert_eq!(dd.stores_dir(), PathBuf::from("/custom/path/stores"));
}
#[test]
fn test_log_file_path() {
fn test_store_paths() {
let dd = DataDir::new("/data");
let path = dd.log_file("abc123");
assert_eq!(path, PathBuf::from("/data/logs/abc123.log"));
let store_id = Uuid::parse_str("a1b2c3d4-e5f6-7890-abcd-ef1234567890").unwrap();
assert_eq!(dd.store_dir(store_id), PathBuf::from("/data/stores/a1b2c3d4-e5f6-7890-abcd-ef1234567890"));
assert_eq!(dd.store_logs_dir(store_id), PathBuf::from("/data/stores/a1b2c3d4-e5f6-7890-abcd-ef1234567890/logs"));
assert_eq!(dd.store_log_file(store_id, "abc123"), PathBuf::from("/data/stores/a1b2c3d4-e5f6-7890-abcd-ef1234567890/logs/abc123.log"));
assert_eq!(dd.store_state_db(store_id), PathBuf::from("/data/stores/a1b2c3d4-e5f6-7890-abcd-ef1234567890/state.db"));
}
#[test]
fn test_default_location_exists() {
// On most systems, default_location should return Some
let location = DataDir::default_location();
// Just verify it doesn't panic - actual path varies by platform
assert!(location.is_some() || true);
}
#[test]
fn test_default_impl() {
let dd = DataDir::default();
// Should either be platform default or ./data fallback
assert!(dd.base().to_str().is_some());
}
}
+21 -7
View File
@@ -1,38 +1,52 @@
//! Lattice Core
//!
//! Core types for the Lattice distributed mesh:
//! - **Node**: Identity with Ed25519 keypair
//! - **NodeIdentity**: Cryptographic identity with Ed25519 keypair
//! - **SigChain**: Append-only cryptographically signed log
//! - **Entry**: Atomic operations in the log
//! - **VectorClock**: Causality tracking for reconciliation
//! - **SyncState**: Per-author sequence tracking for reconciliation
//! - **HLC**: Hybrid Logical Clock for ordering
//! - **Clock**: Time abstraction for testability
//! - **Proto**: Generated protobuf types from lattice.proto
//! - **DataDir**: Platform-specific data directory paths
//! - **SignedEntry**: Entry creation, signing, and verification
//! - **Log**: Append-only log file I/O
//! - **Store**: Persistent KV state from log replay
//! - **CausalIter**: Merge-sort iterator for HLC-ordered sync
pub mod node_identity;
pub mod node;
pub mod sigchain;
pub mod entry;
pub mod vector_clock;
pub mod sync_state;
pub mod hlc;
pub mod clock;
pub mod proto;
pub mod data_dir;
pub mod signed_entry;
pub mod log;
pub mod store;
pub mod meta_store;
pub mod causal_iter;
pub mod store_actor;
// Constants
/// Maximum size of a serialized SignedEntry (16 MB)
pub const MAX_ENTRY_SIZE: usize = 16 * 1024 * 1024;
pub use node::Node;
pub use sigchain::SigChain;
pub use node_identity::{NodeIdentity, PeerStatus};
pub use node::{Node, NodeBuilder, NodeInfo, StoreInfo, StoreHandle, NodeError, NodeEvent, PeerInfo, JoinAcceptance};
pub use sigchain::{SigChain, SigChainManager};
pub use entry::Entry;
pub use vector_clock::VectorClock;
pub use sync_state::{SyncState, AuthorInfo, MissingRange};
pub use hlc::HLC;
pub use clock::{Clock, SystemClock, MockClock};
pub use data_dir::DataDir;
pub use signed_entry::{EntryBuilder, sign_entry, verify_signed_entry, hash_signed_entry};
pub use log::{append_entry, read_entries, LogReader};
pub use log::{append_entry, read_entries, read_entries_after, LogReader};
pub use store::Store;
pub use meta_store::MetaStore;
pub use proto::HeadInfo;
pub use uuid::Uuid;
pub use causal_iter::CausalEntryIter;
pub use store_actor::{StoreActor, StoreCmd, StoreActorError, spawn_store_actor};
+11 -11
View File
@@ -207,7 +207,7 @@ mod tests {
use super::*;
use crate::clock::MockClock;
use crate::hlc::HLC;
use crate::node::Node;
use crate::node_identity::NodeIdentity;
use crate::signed_entry::EntryBuilder;
use std::env::temp_dir;
@@ -226,7 +226,7 @@ mod tests {
let path = temp_log_path("single_v6");
std::fs::remove_file(&path).ok();
let node = Node::generate();
let node = NodeIdentity::generate();
let clock = MockClock::new(1000);
let hlc = HLC::now_with_clock(&clock);
@@ -248,7 +248,7 @@ mod tests {
let path = temp_log_path("multiple_v6");
std::fs::remove_file(&path).ok();
let node = Node::generate();
let node = NodeIdentity::generate();
let clock = MockClock::new(1000);
for i in 1..=5 {
@@ -269,7 +269,7 @@ mod tests {
let path = temp_log_path("after_v6");
std::fs::remove_file(&path).ok();
let node = Node::generate();
let node = NodeIdentity::generate();
let clock = MockClock::new(1000);
let mut entries = Vec::new();
@@ -301,7 +301,7 @@ mod tests {
let path = temp_log_path("not_found_v6");
std::fs::remove_file(&path).ok();
let node = Node::generate();
let node = NodeIdentity::generate();
let clock = MockClock::new(1000);
let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
@@ -321,7 +321,7 @@ mod tests {
let path = temp_log_path("reader_hash_v6");
std::fs::remove_file(&path).ok();
let node = Node::generate();
let node = NodeIdentity::generate();
let clock = MockClock::new(1000);
let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
@@ -368,7 +368,7 @@ mod tests {
let path = temp_log_path("corrupted_v6");
std::fs::remove_file(&path).ok();
let node = Node::generate();
let node = NodeIdentity::generate();
let clock = MockClock::new(1000);
let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
@@ -399,7 +399,7 @@ mod tests {
let path = temp_log_path("truncated_v6");
std::fs::remove_file(&path).ok();
let node = Node::generate();
let node = NodeIdentity::generate();
let clock = MockClock::new(1000);
let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
@@ -430,7 +430,7 @@ mod tests {
let path = temp_log_path("too_large_v6");
std::fs::remove_file(&path).ok();
let node = Node::generate();
let node = NodeIdentity::generate();
let clock = MockClock::new(1000);
// Create payload larger than MAX_ENTRY_SIZE
@@ -455,7 +455,7 @@ mod tests {
let path = temp_log_path("boundary_last_v6");
std::fs::remove_file(&path).ok();
let node = Node::generate();
let node = NodeIdentity::generate();
let clock = MockClock::new(1000);
let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
@@ -506,7 +506,7 @@ mod tests {
let path = temp_log_path("corruption_middle_v6");
std::fs::remove_file(&path).ok();
let node = Node::generate();
let node = NodeIdentity::generate();
let clock = MockClock::new(1000);
// Write 3 entries
+177
View File
@@ -0,0 +1,177 @@
//! MetaStore - global node metadata in meta.db
//!
//! Tables:
//! - stores: UUID → created_at (Unix ms)
//! - meta: "root_store" → UUID (auto-opened on startup)
use redb::{Database, ReadableTable, TableDefinition};
use std::path::Path;
use thiserror::Error;
use uuid::Uuid;
const STORES_TABLE: TableDefinition<&[u8], u64> = TableDefinition::new("stores");
const META_TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("meta");
const META_ROOT_STORE: &str = "root_store";
const META_NAME: &str = "name";
#[derive(Error, Debug)]
pub enum MetaStoreError {
#[error("Database error: {0}")]
Database(#[from] redb::DatabaseError),
#[error("Table error: {0}")]
Table(#[from] redb::TableError),
#[error("Transaction error: {0}")]
Transaction(#[from] redb::TransactionError),
#[error("Commit error: {0}")]
Commit(#[from] redb::CommitError),
#[error("Storage error: {0}")]
Storage(#[from] redb::StorageError),
}
/// Global metadata store
pub struct MetaStore {
db: Database,
}
impl MetaStore {
/// Open or create meta.db at the given path
pub fn open(path: impl AsRef<Path>) -> Result<Self, MetaStoreError> {
let db = Database::create(path)?;
// Ensure tables exist
let write_txn = db.begin_write()?;
{
let _ = write_txn.open_table(STORES_TABLE)?;
let _ = write_txn.open_table(META_TABLE)?;
}
write_txn.commit()?;
Ok(Self { db })
}
/// Register a new store
pub fn add_store(&self, store_id: Uuid) -> Result<(), MetaStoreError> {
let write_txn = self.db.begin_write()?;
{
let mut table = write_txn.open_table(STORES_TABLE)?;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64;
table.insert(store_id.as_bytes().as_slice(), now)?;
}
write_txn.commit()?;
Ok(())
}
/// List all registered stores
pub fn list_stores(&self) -> Result<Vec<Uuid>, MetaStoreError> {
let read_txn = self.db.begin_read()?;
let table = read_txn.open_table(STORES_TABLE)?;
let mut stores = Vec::new();
for result in table.iter()? {
let (key, _created_at) = result?;
let bytes: [u8; 16] = key.value().try_into().unwrap_or([0; 16]);
stores.push(Uuid::from_bytes(bytes));
}
Ok(stores)
}
/// Get the root store ID (auto-opened on startup)
pub fn root_store(&self) -> Result<Option<Uuid>, MetaStoreError> {
let read_txn = self.db.begin_read()?;
let table = read_txn.open_table(META_TABLE)?;
match table.get(META_ROOT_STORE)? {
Some(value) => {
let bytes: [u8; 16] = value.value().try_into().unwrap_or([0; 16]);
Ok(Some(Uuid::from_bytes(bytes)))
}
None => Ok(None),
}
}
/// Set the root store ID
pub fn set_root_store(&self, store_id: Uuid) -> Result<(), MetaStoreError> {
let write_txn = self.db.begin_write()?;
{
let mut table = write_txn.open_table(META_TABLE)?;
table.insert(META_ROOT_STORE, store_id.as_bytes().as_slice())?;
}
write_txn.commit()?;
Ok(())
}
/// Get the node's display name
pub fn name(&self) -> Result<Option<String>, MetaStoreError> {
let read_txn = self.db.begin_read()?;
let table = read_txn.open_table(META_TABLE)?;
match table.get(META_NAME)? {
Some(value) => Ok(Some(String::from_utf8_lossy(value.value()).to_string())),
None => Ok(None),
}
}
/// Set the node's display name
pub fn set_name(&self, name: &str) -> Result<(), MetaStoreError> {
let write_txn = self.db.begin_write()?;
{
let mut table = write_txn.open_table(META_TABLE)?;
table.insert(META_NAME, name.as_bytes())?;
}
write_txn.commit()?;
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::env::temp_dir;
#[test]
fn test_add_and_list_stores() {
let path = temp_dir().join("meta_store_test.db");
let _ = std::fs::remove_file(&path);
let meta = MetaStore::open(&path).unwrap();
let id1 = Uuid::new_v4();
let id2 = Uuid::new_v4();
meta.add_store(id1).unwrap();
meta.add_store(id2).unwrap();
let stores = meta.list_stores().unwrap();
assert_eq!(stores.len(), 2);
assert!(stores.contains(&id1));
assert!(stores.contains(&id2));
let _ = std::fs::remove_file(&path);
}
#[test]
fn test_root_store() {
let path = temp_dir().join("meta_store_root.db");
let _ = std::fs::remove_file(&path);
let meta = MetaStore::open(&path).unwrap();
// Initially no root store
assert_eq!(meta.root_store().unwrap(), None);
let root = Uuid::new_v4();
meta.set_root_store(root).unwrap();
assert_eq!(meta.root_store().unwrap(), Some(root));
let _ = std::fs::remove_file(&path);
}
}
+922 -152
View File
File diff suppressed because it is too large Load Diff
+252
View File
@@ -0,0 +1,252 @@
//! Node identity and cryptographic keys
//!
//! Each node has an Ed25519 keypair:
//! - Private key: stored locally in `identity.key` (never replicated)
//! - Public key: serves as the node's identity (32 bytes)
use ed25519_dalek::{Signature, Signer, SigningKey, Verifier, VerifyingKey};
use rand::rngs::OsRng;
use std::fs;
use std::io::{self, Read, Write};
use std::path::Path;
use thiserror::Error;
/// Errors that can occur during node operations
#[derive(Error, Debug)]
pub enum NodeError {
#[error("IO error: {0}")]
Io(#[from] io::Error),
#[error("Invalid key length: expected 32 bytes, got {0}")]
InvalidKeyLength(usize),
#[error("Invalid signature")]
InvalidSignature,
}
/// A node in the Lattice mesh.
///
/// Each node has an Ed25519 keypair used for signing sigchain entries
/// and establishing trust within the network.
#[derive(Clone)]
pub struct NodeIdentity {
signing_key: SigningKey,
}
impl NodeIdentity {
/// Generate a new node with a random keypair.
pub fn generate() -> Self {
let signing_key = SigningKey::generate(&mut OsRng);
Self { signing_key }
}
/// Create a node from an existing signing key.
pub fn from_signing_key(signing_key: SigningKey) -> Self {
Self { signing_key }
}
/// Load a node's identity from a key file, or generate and save if it doesn't exist.
pub fn load_or_generate(path: impl AsRef<Path>) -> Result<Self, NodeError> {
let path = path.as_ref();
if path.exists() {
Self::load(path)
} else {
let node = Self::generate();
node.save(path)?;
Ok(node)
}
}
/// Load a node's identity from a key file.
pub fn load(path: impl AsRef<Path>) -> Result<Self, NodeError> {
let mut file = fs::File::open(path)?;
let mut bytes = Vec::new();
file.read_to_end(&mut bytes)?;
if bytes.len() != 32 {
return Err(NodeError::InvalidKeyLength(bytes.len()));
}
let key_bytes: [u8; 32] = bytes.try_into().unwrap();
let signing_key = SigningKey::from_bytes(&key_bytes);
Ok(Self { signing_key })
}
/// Save the node's private key to a file.
pub fn save(&self, path: impl AsRef<Path>) -> Result<(), NodeError> {
let path = path.as_ref();
// Create parent directories if they don't exist
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)?;
}
let mut file = fs::File::create(path)?;
file.write_all(self.signing_key.as_bytes())?;
Ok(())
}
/// Get the node's public key (identity).
pub fn public_key(&self) -> VerifyingKey {
self.signing_key.verifying_key()
}
/// Get the node's public key as bytes (32 bytes).
pub fn public_key_bytes(&self) -> [u8; 32] {
self.signing_key.verifying_key().to_bytes()
}
/// Get the signing key for creating signatures.
pub fn signing_key(&self) -> &SigningKey {
&self.signing_key
}
/// Get the secret key bytes (32 bytes) for Iroh integration.
/// WARNING: Handle with care - this exposes the private key material.
pub fn secret_key_bytes(&self) -> [u8; 32] {
self.signing_key.to_bytes()
}
/// Sign a message.
pub fn sign(&self, message: &[u8]) -> Signature {
self.signing_key.sign(message)
}
/// Verify a signature against this node's public key.
pub fn verify(&self, message: &[u8], signature: &Signature) -> Result<(), NodeError> {
self.public_key()
.verify(message, signature)
.map_err(|_| NodeError::InvalidSignature)
}
/// Verify a signature using a raw public key.
pub fn verify_with_key(
public_key: &VerifyingKey,
message: &[u8],
signature: &Signature,
) -> Result<(), NodeError> {
public_key
.verify(message, signature)
.map_err(|_| NodeError::InvalidSignature)
}
}
/// Peer status values used across the system
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum PeerStatus {
/// Peer has been invited but hasn't joined yet
Invited,
/// Peer is active and can sync
Active,
/// Peer is temporarily inactive
Dormant,
}
impl PeerStatus {
pub fn as_str(&self) -> &'static str {
match self {
PeerStatus::Invited => "invited",
PeerStatus::Active => "active",
PeerStatus::Dormant => "dormant",
}
}
pub fn from_str(s: &str) -> Option<PeerStatus> {
match s {
"invited" => Some(PeerStatus::Invited),
"active" => Some(PeerStatus::Active),
"dormant" => Some(PeerStatus::Dormant),
_ => None,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::env::temp_dir;
#[test]
fn test_generate() {
let node = NodeIdentity::generate();
let pk = node.public_key_bytes();
assert_eq!(pk.len(), 32);
}
#[test]
fn test_sign_and_verify() {
let node = NodeIdentity::generate();
let message = b"hello lattice";
let signature = node.sign(message);
assert!(node.verify(message, &signature).is_ok());
}
#[test]
fn test_verify_wrong_message() {
let node = NodeIdentity::generate();
let signature = node.sign(b"original");
assert!(node.verify(b"tampered", &signature).is_err());
}
#[test]
fn test_verify_with_different_key() {
let node1 = NodeIdentity::generate();
let node2 = NodeIdentity::generate();
let signature = node1.sign(b"message");
assert!(node2.verify(b"message", &signature).is_err());
}
#[test]
fn test_save_and_load() {
let temp_path = temp_dir().join("lattice_test_identity.key");
// Generate and save
let node1 = NodeIdentity::generate();
let pk1 = node1.public_key_bytes();
node1.save(&temp_path).unwrap();
// Load and verify same key
let node2 = NodeIdentity::load(&temp_path).unwrap();
let pk2 = node2.public_key_bytes();
assert_eq!(pk1, pk2);
// Cleanup
fs::remove_file(&temp_path).ok();
}
#[test]
fn test_load_or_generate() {
let temp_path = temp_dir().join("lattice_test_identity2.key");
// Remove if exists
fs::remove_file(&temp_path).ok();
// First call: generates
let node1 = NodeIdentity::load_or_generate(&temp_path).unwrap();
let pk1 = node1.public_key_bytes();
// Second call: loads existing
let node2 = NodeIdentity::load_or_generate(&temp_path).unwrap();
let pk2 = node2.public_key_bytes();
assert_eq!(pk1, pk2);
// Cleanup
fs::remove_file(&temp_path).ok();
}
#[test]
fn test_verify_with_key_static() {
let node = NodeIdentity::generate();
let pk = node.public_key();
let message = b"test message";
let signature = node.sign(message);
assert!(NodeIdentity::verify_with_key(&pk, message, &signature).is_ok());
}
}
+3 -1
View File
@@ -31,7 +31,9 @@ mod tests {
fn test_entry_with_ops() {
let entry = Entry {
version: 1,
store_id: vec![1u8; 16],
prev_hash: vec![0u8; 32],
parent_hashes: vec![],
seq: 5,
timestamp: Some(Hlc {
wall_time: 1000,
@@ -40,7 +42,7 @@ mod tests {
ops: vec![
Operation {
op_type: Some(operation::OpType::Put(PutOp {
key: "/nodes/abc".to_string(),
key: b"/nodes/abc".to_vec(),
value: b"hello".to_vec(),
})),
},
+188 -25
View File
@@ -4,7 +4,7 @@
//! before appending (correct seq, prev_hash, valid signature) and persists to disk.
use crate::log::{append_entry, read_entries, LogError};
use crate::node::Node;
use crate::node_identity::NodeIdentity;
use crate::proto::{Entry, SignedEntry};
use crate::signed_entry::{hash_signed_entry, verify_signed_entry};
use prost::Message;
@@ -23,6 +23,9 @@ pub enum SigChainError {
#[error("Wrong author: expected {expected}, got {got}")]
WrongAuthor { expected: String, got: String },
#[error("Wrong store_id: expected {expected}, got {got}")]
WrongStoreId { expected: String, got: String },
#[error("Invalid sequence: expected {expected}, got {got}")]
InvalidSequence { expected: u64, got: u64 },
@@ -34,11 +37,14 @@ pub enum SigChainError {
}
/// An append-only log where each entry is cryptographically signed
/// and hash-linked to the previous entry.
/// and hash-linked to the previous entry, scoped to a specific store.
pub struct SigChain {
/// Path to the log file
log_path: PathBuf,
/// Store UUID (16 bytes)
store_id: [u8; 16],
/// Author's public key (32 bytes)
author_id: [u8; 32],
@@ -50,10 +56,11 @@ pub struct SigChain {
}
impl SigChain {
/// Create a new empty sigchain for an author
pub fn new(log_path: impl AsRef<Path>, author_id: [u8; 32]) -> Self {
/// Create a new empty sigchain for a (store, author) pair
pub fn new(log_path: impl AsRef<Path>, store_id: [u8; 16], author_id: [u8; 32]) -> Self {
Self {
log_path: log_path.as_ref().to_path_buf(),
store_id,
author_id,
next_seq: 1,
last_hash: [0u8; 32],
@@ -61,11 +68,11 @@ impl SigChain {
}
/// Load a sigchain from an existing log file
pub fn from_log(log_path: impl AsRef<Path>, author_id: [u8; 32]) -> Result<Self, SigChainError> {
pub fn from_log(log_path: impl AsRef<Path>, store_id: [u8; 16], author_id: [u8; 32]) -> Result<Self, SigChainError> {
let log_path = log_path.as_ref().to_path_buf();
let entries = read_entries(&log_path)?;
let mut chain = Self::new(&log_path, author_id);
let mut chain = Self::new(&log_path, store_id, author_id);
for signed_entry in entries {
// Verify signature
@@ -85,6 +92,18 @@ impl SigChain {
// Decode Entry
let entry = Entry::decode(&signed_entry.entry_bytes[..])?;
// Validate store_id
// Note: Empty/malformed store_id becomes [0u8;16], which fails validation
// against any real UUID store. This intentionally rejects legacy entries.
let entry_store: [u8; 16] = entry.store_id.clone().try_into()
.unwrap_or([0u8; 16]);
if entry_store != store_id {
return Err(SigChainError::WrongStoreId {
expected: hex::encode(store_id),
got: hex::encode(entry_store),
});
}
// Validate sequence
if entry.seq != chain.next_seq {
return Err(SigChainError::InvalidSequence {
@@ -127,6 +146,11 @@ impl SigChain {
&self.last_hash
}
/// Get the log file path
pub fn log_path(&self) -> &std::path::Path {
&self.log_path
}
/// Get the current length of the chain
pub fn len(&self) -> u64 {
self.next_seq - 1
@@ -156,6 +180,18 @@ impl SigChain {
// Decode entry
let entry = Entry::decode(&signed_entry.entry_bytes[..])?;
// Validate store_id
// Note: Empty/malformed store_id becomes [0u8;16], which fails validation
// against any real UUID store. This intentionally rejects legacy entries.
let entry_store: [u8; 16] = entry.store_id.clone().try_into()
.unwrap_or([0u8; 16]);
if entry_store != self.store_id {
return Err(SigChainError::WrongStoreId {
expected: hex::encode(self.store_id),
got: hex::encode(entry_store),
});
}
// Validate sequence
if entry.seq != self.next_seq {
return Err(SigChainError::InvalidSequence {
@@ -193,7 +229,7 @@ impl SigChain {
}
/// Create and append a new entry using the node's key
pub fn create_entry(&mut self, node: &Node, ops: Vec<crate::proto::Operation>) -> Result<SignedEntry, SigChainError> {
pub fn create_entry(&mut self, node: &NodeIdentity, ops: Vec<crate::proto::Operation>) -> Result<SignedEntry, SigChainError> {
use crate::clock::SystemClock;
use crate::hlc::HLC;
use crate::signed_entry::EntryBuilder;
@@ -201,6 +237,7 @@ impl SigChain {
let hlc = HLC::now_with_clock(&SystemClock);
let mut builder = EntryBuilder::new(self.next_seq, hlc)
.store_id(self.store_id.to_vec())
.prev_hash(self.last_hash.to_vec());
// Add operations
@@ -216,12 +253,96 @@ impl SigChain {
}
}
/// Manages multiple SigChains (one per author) for a store.
/// Provides unified interface for appending entries from any author.
pub struct SigChainManager {
/// Directory containing log files (one per author)
logs_dir: PathBuf,
/// Store UUID (16 bytes)
store_id: [u8; 16],
/// Cache of loaded SigChains by author
chains: std::collections::HashMap<[u8; 32], SigChain>,
}
impl SigChainManager {
/// Create a new manager for a store's logs directory
pub fn new(logs_dir: impl AsRef<Path>, store_id: [u8; 16]) -> Self {
Self {
logs_dir: logs_dir.as_ref().to_path_buf(),
store_id,
chains: std::collections::HashMap::new(),
}
}
/// Get or create a SigChain for an author
pub fn get_or_create(&mut self, author: [u8; 32]) -> &mut SigChain {
self.chains.entry(author).or_insert_with(|| {
let author_hex = hex::encode(author);
let log_path = self.logs_dir.join(format!("{}.log", author_hex));
// Try to load existing log, or create new
SigChain::from_log(&log_path, self.store_id, author)
.unwrap_or_else(|_| SigChain::new(&log_path, self.store_id, author))
})
}
/// Get the local node's sigchain (for creating new entries)
pub fn get(&self, author: &[u8; 32]) -> Option<&SigChain> {
self.chains.get(author)
}
/// Append an entry to the appropriate author's log
/// This is the main entry point for all entry writes (from put, sync, etc.)
pub fn append_entry(&mut self, entry: &SignedEntry) -> Result<(), SigChainError> {
let author: [u8; 32] = entry.author_id.clone()
.try_into()
.map_err(|_| SigChainError::WrongAuthor {
expected: "32 bytes".to_string(),
got: format!("{} bytes", entry.author_id.len()),
})?;
// For synced entries, we can't validate seq/prev_hash since they may arrive
// out of order. Just append to the log file directly.
let chain = self.get_or_create(author);
append_entry(chain.log_path(), entry)?;
Ok(())
}
/// Get the logs directory path
pub fn logs_dir(&self) -> &Path {
&self.logs_dir
}
/// Get log directory statistics (file count, total bytes)
pub fn log_stats(&self) -> (usize, u64) {
if !self.logs_dir.exists() {
return (0, 0);
}
let mut total_size = 0u64;
let mut file_count = 0;
if let Ok(entries) = std::fs::read_dir(&self.logs_dir) {
for entry in entries.flatten() {
if let Ok(meta) = entry.metadata() {
if meta.is_file() {
total_size += meta.len();
file_count += 1;
}
}
}
}
(file_count, total_size)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::clock::MockClock;
use crate::hlc::HLC;
use crate::node::Node;
use crate::node_identity::NodeIdentity;
use crate::proto::{operation, Operation, PutOp};
use crate::signed_entry::EntryBuilder;
use std::env::temp_dir;
@@ -230,12 +351,14 @@ mod tests {
temp_dir().join(format!("lattice_sigchain_test_{}.log", name))
}
const TEST_STORE: [u8; 16] = [1u8; 16];
#[test]
fn test_new_sigchain() {
let path = temp_log_path("new");
let author = [1u8; 32];
let chain = SigChain::new(&path, author);
let chain = SigChain::new(&path, TEST_STORE, author);
assert_eq!(chain.author_id(), &author);
assert_eq!(chain.next_seq(), 1);
@@ -249,12 +372,13 @@ mod tests {
let path = temp_log_path("append");
std::fs::remove_file(&path).ok();
let node = Node::generate();
let node = NodeIdentity::generate();
let author = node.public_key_bytes();
let mut chain = SigChain::new(&path, author);
let mut chain = SigChain::new(&path, TEST_STORE, author);
let clock = MockClock::new(1000);
let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
.store_id(TEST_STORE.to_vec())
.prev_hash([0u8; 32].to_vec())
.put("/key", b"value".to_vec())
.sign(&node);
@@ -273,13 +397,14 @@ mod tests {
let path = temp_log_path("multiple");
std::fs::remove_file(&path).ok();
let node = Node::generate();
let node = NodeIdentity::generate();
let author = node.public_key_bytes();
let mut chain = SigChain::new(&path, author);
let mut chain = SigChain::new(&path, TEST_STORE, author);
let clock = MockClock::new(1000);
for i in 1..=3 {
let entry = EntryBuilder::new(i, HLC::now_with_clock(&clock))
.store_id(TEST_STORE.to_vec())
.prev_hash(chain.last_hash.to_vec())
.put(format!("/key/{}", i), format!("value{}", i).into_bytes())
.sign(&node);
@@ -297,15 +422,16 @@ mod tests {
let path = temp_log_path("from_log");
std::fs::remove_file(&path).ok();
let node = Node::generate();
let node = NodeIdentity::generate();
let author = node.public_key_bytes();
let clock = MockClock::new(1000);
// Write some entries
{
let mut chain = SigChain::new(&path, author);
let mut chain = SigChain::new(&path, TEST_STORE, author);
for i in 1..=3 {
let entry = EntryBuilder::new(i, HLC::now_with_clock(&clock))
.store_id(TEST_STORE.to_vec())
.prev_hash(chain.last_hash.to_vec())
.put("/key", b"val".to_vec())
.sign(&node);
@@ -314,7 +440,7 @@ mod tests {
}
// Reload from log
let chain = SigChain::from_log(&path, author).unwrap();
let chain = SigChain::from_log(&path, TEST_STORE, author).unwrap();
assert_eq!(chain.len(), 3);
assert_eq!(chain.next_seq(), 4);
@@ -327,13 +453,14 @@ mod tests {
let path = temp_log_path("wrong_seq");
std::fs::remove_file(&path).ok();
let node = Node::generate();
let node = NodeIdentity::generate();
let author = node.public_key_bytes();
let mut chain = SigChain::new(&path, author);
let mut chain = SigChain::new(&path, TEST_STORE, author);
let clock = MockClock::new(1000);
// Try to append with wrong seq (2 instead of 1)
let entry = EntryBuilder::new(2, HLC::now_with_clock(&clock))
.store_id(TEST_STORE.to_vec())
.prev_hash([0u8; 32].to_vec())
.put("/key", b"val".to_vec())
.sign(&node);
@@ -350,13 +477,14 @@ mod tests {
let path = temp_log_path("wrong_prev");
std::fs::remove_file(&path).ok();
let node = Node::generate();
let node = NodeIdentity::generate();
let author = node.public_key_bytes();
let mut chain = SigChain::new(&path, author);
let mut chain = SigChain::new(&path, TEST_STORE, author);
let clock = MockClock::new(1000);
// First entry
let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock))
.store_id(TEST_STORE.to_vec())
.prev_hash([0u8; 32].to_vec())
.put("/key", b"v1".to_vec())
.sign(&node);
@@ -364,6 +492,7 @@ mod tests {
// Second entry with wrong prev_hash
let entry2 = EntryBuilder::new(2, HLC::now_with_clock(&clock))
.store_id(TEST_STORE.to_vec())
.prev_hash([99u8; 32].to_vec()) // Wrong!
.put("/key", b"v2".to_vec())
.sign(&node);
@@ -380,13 +509,14 @@ mod tests {
let path = temp_log_path("wrong_author");
std::fs::remove_file(&path).ok();
let node = Node::generate();
let node = NodeIdentity::generate();
let other_author = [99u8; 32]; // Different author
let mut chain = SigChain::new(&path, other_author);
let mut chain = SigChain::new(&path, TEST_STORE, other_author);
let clock = MockClock::new(1000);
// Entry signed by node but chain expects other_author
let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
.store_id(TEST_STORE.to_vec())
.prev_hash([0u8; 32].to_vec())
.put("/key", b"val".to_vec())
.sign(&node);
@@ -403,14 +533,14 @@ mod tests {
let path = temp_log_path("create");
std::fs::remove_file(&path).ok();
let node = Node::generate();
let node = NodeIdentity::generate();
let author = node.public_key_bytes();
let mut chain = SigChain::new(&path, author);
let mut chain = SigChain::new(&path, TEST_STORE, author);
let ops = vec![
Operation {
op_type: Some(operation::OpType::Put(PutOp {
key: "/test".to_string(),
key: b"/test".to_vec(),
value: b"hello".to_vec(),
})),
},
@@ -427,4 +557,37 @@ mod tests {
std::fs::remove_file(&path).ok();
}
#[test]
fn test_reject_wrong_store_id() {
let path_a = temp_log_path("storeid_a");
let path_b = temp_log_path("storeid_b");
std::fs::remove_file(&path_a).ok();
std::fs::remove_file(&path_b).ok();
let node = NodeIdentity::generate();
let author = node.public_key_bytes();
let clock = MockClock::new(1000);
let store_a = [0xAAu8; 16];
let store_b = [0xBBu8; 16];
// Create valid entry for store A
let mut chain_a = SigChain::new(&path_a, store_a, author);
let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
.store_id(store_a.to_vec())
.prev_hash([0u8; 32].to_vec())
.put("/key", b"val".to_vec())
.sign(&node);
chain_a.append(&entry).unwrap();
// Try to replay that entry into store B's chain
let mut chain_b = SigChain::new(&path_b, store_b, author);
let result = chain_b.append(&entry);
assert!(matches!(result, Err(SigChainError::WrongStoreId { .. })));
std::fs::remove_file(&path_a).ok();
std::fs::remove_file(&path_b).ok();
}
}
+32 -14
View File
@@ -7,7 +7,7 @@
//! - Computing entry hashes for prev_hash linking
use crate::hlc::HLC;
use crate::node::{Node, NodeError};
use crate::node_identity::{NodeIdentity, NodeError};
use crate::proto::{Entry, Hlc, Operation, PutOp, DeleteOp, SignedEntry, operation};
use ed25519_dalek::{Signature, VerifyingKey};
use prost::Message;
@@ -32,7 +32,9 @@ pub enum EntryError {
/// Builder for creating Entry messages
pub struct EntryBuilder {
version: u32,
store_id: Vec<u8>,
prev_hash: Vec<u8>,
parent_hashes: Vec<Vec<u8>>,
seq: u64,
timestamp: HLC,
ops: Vec<Operation>,
@@ -43,21 +45,35 @@ impl EntryBuilder {
pub fn new(seq: u64, timestamp: HLC) -> Self {
Self {
version: 1,
prev_hash: vec![0u8; 32], // Genesis or will be set
store_id: Vec::new(),
prev_hash: vec![0u8; 32],
parent_hashes: Vec::new(),
seq,
timestamp,
ops: Vec::new(),
}
}
/// Set the previous entry hash (for chaining)
/// Set the store ID (16-byte UUID)
pub fn store_id(mut self, id: impl Into<Vec<u8>>) -> Self {
self.store_id = id.into();
self
}
/// Set the previous entry hash (for sigchain linking)
pub fn prev_hash(mut self, hash: impl Into<Vec<u8>>) -> Self {
self.prev_hash = hash.into();
self
}
/// Set the parent hashes (for DAG ancestry)
pub fn parent_hashes(mut self, hashes: Vec<Vec<u8>>) -> Self {
self.parent_hashes = hashes;
self
}
/// Add a Put operation
pub fn put(mut self, key: impl Into<String>, value: impl Into<Vec<u8>>) -> Self {
pub fn put(mut self, key: impl Into<Vec<u8>>, value: impl Into<Vec<u8>>) -> Self {
self.ops.push(Operation {
op_type: Some(operation::OpType::Put(PutOp {
key: key.into(),
@@ -68,7 +84,7 @@ impl EntryBuilder {
}
/// Add a Delete operation
pub fn delete(mut self, key: impl Into<String>) -> Self {
pub fn delete(mut self, key: impl Into<Vec<u8>>) -> Self {
self.ops.push(Operation {
op_type: Some(operation::OpType::Delete(DeleteOp {
key: key.into(),
@@ -87,7 +103,9 @@ impl EntryBuilder {
pub fn build(self) -> Entry {
Entry {
version: self.version,
store_id: self.store_id,
prev_hash: self.prev_hash,
parent_hashes: self.parent_hashes,
seq: self.seq,
timestamp: Some(Hlc {
wall_time: self.timestamp.wall_time,
@@ -98,14 +116,14 @@ impl EntryBuilder {
}
/// Build and sign the entry, returning a SignedEntry
pub fn sign(self, node: &Node) -> SignedEntry {
pub fn sign(self, node: &NodeIdentity) -> SignedEntry {
let entry = self.build();
sign_entry(&entry, node)
}
}
/// Sign an Entry to create a SignedEntry
pub fn sign_entry(entry: &Entry, node: &Node) -> SignedEntry {
pub fn sign_entry(entry: &Entry, node: &NodeIdentity) -> SignedEntry {
let entry_bytes = entry.encode_to_vec();
let signature = node.sign(&entry_bytes);
@@ -134,7 +152,7 @@ pub fn verify_signed_entry(signed: &SignedEntry) -> Result<Entry, EntryError> {
let signature = Signature::from_bytes(&sig_bytes);
// Verify
Node::verify_with_key(&public_key, &signed.entry_bytes, &signature)?;
NodeIdentity::verify_with_key(&public_key, &signed.entry_bytes, &signature)?;
// Decode entry
let entry = Entry::decode(&signed.entry_bytes[..])?;
@@ -174,7 +192,7 @@ mod tests {
#[test]
fn test_sign_and_verify() {
let node = Node::generate();
let node = NodeIdentity::generate();
let clock = MockClock::new(1000);
let hlc = HLC::now_with_clock(&clock);
@@ -193,7 +211,7 @@ mod tests {
#[test]
fn test_verify_tampered_fails() {
let node = Node::generate();
let node = NodeIdentity::generate();
let clock = MockClock::new(1000);
let hlc = HLC::now_with_clock(&clock);
@@ -209,8 +227,8 @@ mod tests {
#[test]
fn test_verify_wrong_key_fails() {
let node1 = Node::generate();
let node2 = Node::generate();
let node1 = NodeIdentity::generate();
let node2 = NodeIdentity::generate();
let clock = MockClock::new(1000);
let hlc = HLC::now_with_clock(&clock);
@@ -226,7 +244,7 @@ mod tests {
#[test]
fn test_hash_signed_entry() {
let node = Node::generate();
let node = NodeIdentity::generate();
let clock = MockClock::new(1000);
let hlc = HLC::now_with_clock(&clock);
@@ -244,7 +262,7 @@ mod tests {
#[test]
fn test_prev_hash_chaining() {
let node = Node::generate();
let node = NodeIdentity::generate();
let clock = MockClock::new(1000);
// First entry
File diff suppressed because it is too large Load Diff
+299
View File
@@ -0,0 +1,299 @@
//! Store Actor - dedicated thread that owns Store and processes commands via channel
use crate::{
EntryBuilder, HeadInfo, NodeIdentity, SigChain, SigChainManager, Store, Uuid,
hlc::HLC,
proto::AuthorState,
sigchain::SigChainError,
store::StoreError,
sync_state::SyncState,
proto::SignedEntry,
log,
};
use tokio::sync::{mpsc, oneshot, broadcast};
use std::thread::{self, JoinHandle};
/// Commands sent to the store actor
pub enum StoreCmd {
Get {
key: Vec<u8>,
resp: oneshot::Sender<Result<Option<Vec<u8>>, StoreError>>,
},
GetHeads {
key: Vec<u8>,
resp: oneshot::Sender<Result<Vec<HeadInfo>, StoreError>>,
},
List {
include_deleted: bool,
resp: oneshot::Sender<Result<Vec<(Vec<u8>, Vec<u8>)>, StoreError>>,
},
ListByPrefix {
prefix: Vec<u8>,
include_deleted: bool,
resp: oneshot::Sender<Result<Vec<(Vec<u8>, Vec<u8>)>, StoreError>>,
},
Put {
key: Vec<u8>,
value: Vec<u8>,
resp: oneshot::Sender<Result<u64, StoreActorError>>,
},
Delete {
key: Vec<u8>,
resp: oneshot::Sender<Result<u64, StoreActorError>>,
},
LogSeq {
resp: oneshot::Sender<u64>,
},
AppliedSeq {
resp: oneshot::Sender<Result<u64, StoreError>>,
},
AuthorState {
author: [u8; 32],
resp: oneshot::Sender<Result<Option<AuthorState>, StoreError>>,
},
// Sync-related commands
SyncState {
resp: oneshot::Sender<Result<SyncState, StoreError>>,
},
ReadEntriesAfter {
author: [u8; 32],
from_hash: Option<[u8; 32]>,
resp: oneshot::Sender<Result<Vec<SignedEntry>, StoreError>>,
},
ApplyEntry {
entry: SignedEntry,
resp: oneshot::Sender<Result<(), StoreError>>,
},
LogStats {
resp: oneshot::Sender<(usize, u64)>,
},
Shutdown,
}
#[derive(Debug)]
pub enum StoreActorError {
Store(StoreError),
SigChain(SigChainError),
}
impl From<StoreError> for StoreActorError {
fn from(e: StoreError) -> Self {
StoreActorError::Store(e)
}
}
impl From<SigChainError> for StoreActorError {
fn from(e: SigChainError) -> Self {
StoreActorError::SigChain(e)
}
}
impl std::fmt::Display for StoreActorError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
StoreActorError::Store(e) => write!(f, "Store error: {}", e),
StoreActorError::SigChain(e) => write!(f, "SigChain error: {}", e),
}
}
}
impl std::error::Error for StoreActorError {}
/// The store actor - runs in its own thread, owns Store and SigChainManager
pub struct StoreActor {
store_id: Uuid,
store: Store,
chain_manager: SigChainManager,
node: NodeIdentity,
rx: mpsc::Receiver<StoreCmd>,
/// Broadcast sender for emitting entries after they're committed locally
entry_tx: broadcast::Sender<SignedEntry>,
}
impl StoreActor {
/// Create a new store actor (but don't start the thread yet)
pub fn new(
store_id: Uuid,
store: Store,
sigchain: SigChain,
node: NodeIdentity,
rx: mpsc::Receiver<StoreCmd>,
entry_tx: broadcast::Sender<SignedEntry>,
) -> Self {
// Derive logs_dir from sigchain's log file path
let logs_dir = sigchain.log_path()
.parent()
.map(|p| p.to_path_buf())
.unwrap_or_default();
// Create chain manager and register the local node's sigchain
let mut chain_manager = SigChainManager::new(&logs_dir, *store_id.as_bytes());
let local_author = node.public_key_bytes();
chain_manager.get_or_create(local_author); // Pre-initialize local chain
Self {
store_id,
store,
chain_manager,
node,
rx,
entry_tx,
}
}
/// Run the actor loop - processes commands until Shutdown received
/// Uses blocking_recv since redb is sync and we run in spawn_blocking
pub fn run(mut self) {
while let Some(cmd) = self.rx.blocking_recv() {
match cmd {
StoreCmd::Get { key, resp } => {
let _ = resp.send(self.store.get(&key));
}
StoreCmd::GetHeads { key, resp } => {
let _ = resp.send(self.store.get_heads(&key));
}
StoreCmd::List { include_deleted, resp } => {
let _ = resp.send(self.store.list_all(include_deleted));
}
StoreCmd::ListByPrefix { prefix, include_deleted, resp } => {
let _ = resp.send(self.store.list_by_prefix(&prefix, include_deleted));
}
StoreCmd::Put { key, value, resp } => {
let result = self.do_put(&key, &value);
let _ = resp.send(result);
}
StoreCmd::Delete { key, resp } => {
let result = self.do_delete(&key);
let _ = resp.send(result);
}
StoreCmd::LogSeq { resp } => {
let local_author = self.node.public_key_bytes();
let len = self.chain_manager.get(&local_author)
.map(|c| c.len())
.unwrap_or(0);
let _ = resp.send(len);
}
StoreCmd::AppliedSeq { resp } => {
let author = self.node.public_key_bytes();
let result = self.store.author_state(&author)
.map(|s| s.map(|a| a.seq).unwrap_or(0));
let _ = resp.send(result);
}
StoreCmd::AuthorState { author, resp } => {
let _ = resp.send(self.store.author_state(&author));
}
StoreCmd::SyncState { resp } => {
let _ = resp.send(self.store.sync_state());
}
StoreCmd::ReadEntriesAfter { author, from_hash, resp } => {
// Read entries from the log file for this author
let result = self.do_read_entries_after(&author, from_hash);
let _ = resp.send(result);
}
StoreCmd::ApplyEntry { entry, resp } => {
// Use SigChainManager to append to the correct author's log
if let Err(e) = self.chain_manager.append_entry(&entry) {
let _ = resp.send(Err(StoreError::from(e)));
continue;
}
// Then apply to store
let result = self.store.apply_entry(&entry);
let _ = resp.send(result);
}
StoreCmd::LogStats { resp } => {
let _ = resp.send(self.chain_manager.log_stats());
}
StoreCmd::Shutdown => {
break;
}
}
}
}
fn do_put(&mut self, key: &[u8], value: &[u8]) -> Result<u64, StoreActorError> {
let heads = self.store.get_heads(key)?;
// Idempotency check (pure function)
if !Store::needs_put(&heads, value) {
let local_author = self.node.public_key_bytes();
return Ok(self.chain_manager.get(&local_author).map(|c| c.len()).unwrap_or(0));
}
let parent_hashes: Vec<Vec<u8>> = heads.iter().map(|h| h.hash.clone()).collect();
self.commit_entry(parent_hashes, |b| b.put(key.to_vec(), value.to_vec()))
}
fn do_delete(&mut self, key: &[u8]) -> Result<u64, StoreActorError> {
let heads = self.store.get_heads(key)?;
// Idempotency check (pure function)
if !Store::needs_delete(&heads) {
let local_author = self.node.public_key_bytes();
return Ok(self.chain_manager.get(&local_author).map(|c| c.len()).unwrap_or(0));
}
let parent_hashes: Vec<Vec<u8>> = heads.iter().map(|h| h.hash.clone()).collect();
self.commit_entry(parent_hashes, |b| b.delete(key.to_vec()))
}
fn commit_entry<F>(&mut self, parent_hashes: Vec<Vec<u8>>, build: F) -> Result<u64, StoreActorError>
where
F: FnOnce(EntryBuilder) -> EntryBuilder,
{
let local_author = self.node.public_key_bytes();
let sigchain = self.chain_manager.get_or_create(local_author);
let seq = sigchain.len() + 1;
let prev_hash = *sigchain.last_hash();
let builder = EntryBuilder::new(seq, HLC::now())
.store_id(self.store_id.as_bytes().to_vec())
.prev_hash(prev_hash.to_vec())
.parent_hashes(parent_hashes);
let entry = build(builder).sign(&self.node);
// Append to local sigchain
let sigchain = self.chain_manager.get_or_create(local_author);
sigchain.append(&entry)?;
self.store.apply_entry(&entry)?;
// Broadcast the entry to listeners (for gossip)
let _ = self.entry_tx.send(entry.clone());
Ok(seq)
}
fn do_read_entries_after(
&self,
author: &[u8; 32],
from_hash: Option<[u8; 32]>,
) -> Result<Vec<SignedEntry>, StoreError> {
// Build log path for this author
let author_hex = hex::encode(author);
let log_path = self.chain_manager.logs_dir().join(format!("{}.log", author_hex));
if !log_path.exists() {
return Ok(Vec::new()); // No log file for this author
}
// Use lattice_core's read_entries_after
log::read_entries_after(&log_path, from_hash)
.map_err(StoreError::from)
}
}
/// Spawn a store actor in a new thread, returns (cmd_tx, entry_tx, join_handle)
/// Uses std::thread since redb is blocking
pub fn spawn_store_actor(
store_id: Uuid,
store: Store,
sigchain: SigChain,
node: NodeIdentity,
) -> (mpsc::Sender<StoreCmd>, broadcast::Sender<SignedEntry>, JoinHandle<()>) {
let (tx, rx) = mpsc::channel(32);
let (entry_tx, _entry_rx) = broadcast::channel(64);
let actor = StoreActor::new(store_id, store, sigchain, node, rx, entry_tx.clone());
let handle = thread::spawn(move || actor.run());
(tx, entry_tx, handle)
}
+325
View File
@@ -0,0 +1,325 @@
//! Sync state for causality tracking and reconciliation
use std::collections::{HashMap, HashSet};
/// Author ID type (32-byte Ed25519 public key)
pub type Author = [u8; 32];
/// Per-author sync information: seq + all head hashes.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AuthorInfo {
pub seq: u64,
pub heads: HashSet<[u8; 32]>, // All head hashes for this author
}
impl AuthorInfo {
pub fn new(seq: u64, hash: [u8; 32]) -> Self {
let mut heads = HashSet::new();
heads.insert(hash);
Self { seq, heads }
}
pub fn with_heads(seq: u64, heads: HashSet<[u8; 32]>) -> Self {
Self { seq, heads }
}
}
/// Sync state tracking per-author sequence numbers and head hashes.
///
/// Used during reconciliation to identify missing entries between peers.
/// Tracks all head hashes per author to handle forks correctly.
#[derive(Debug, Clone, Default)]
pub struct SyncState {
authors: HashMap<Author, AuthorInfo>,
}
/// Describes entries needed from a peer for a specific author.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MissingRange {
pub author: Author,
pub from_seq: u64, // exclusive - we have up to this
pub from_hash: [u8; 32], // hash to resume reading after (zero = start)
pub to_seq: u64, // inclusive - peer has up to this
}
impl SyncState {
/// Create a new empty sync state.
pub fn new() -> Self {
Self {
authors: HashMap::new(),
}
}
/// Get the info for an author (returns None if not present).
pub fn get(&self, author: &Author) -> Option<&AuthorInfo> {
self.authors.get(author)
}
/// Get the sequence number for an author (returns 0 if not present).
pub fn seq(&self, author: &Author) -> u64 {
self.authors.get(author).map(|i| i.seq).unwrap_or(0)
}
/// Get head hashes for an author (returns empty set if not present).
pub fn heads(&self, author: &Author) -> HashSet<[u8; 32]> {
self.authors.get(author).map(|i| i.heads.clone()).unwrap_or_default()
}
/// Set the info for an author (single hash convenience method).
pub fn set(&mut self, author: Author, seq: u64, hash: [u8; 32]) {
self.authors.insert(author, AuthorInfo::new(seq, hash));
}
/// Set the info for an author with multiple heads.
pub fn set_heads(&mut self, author: Author, seq: u64, heads: HashSet<[u8; 32]>) {
self.authors.insert(author, AuthorInfo::with_heads(seq, heads));
}
/// Add a head hash for an author (updates seq if higher).
pub fn add_head(&mut self, author: Author, seq: u64, hash: [u8; 32]) {
if let Some(info) = self.authors.get_mut(&author) {
info.heads.insert(hash);
if seq > info.seq {
info.seq = seq;
}
} else {
self.set(author, seq, hash);
}
}
/// Get all authors and their info.
pub fn authors(&self) -> &HashMap<Author, AuthorInfo> {
&self.authors
}
/// Compute what entries we're missing compared to a peer's state.
///
/// Returns ranges of entries we need from the peer.
/// Compares hash sets when seq matches to detect forks.
pub fn diff(&self, peer: &SyncState) -> Vec<MissingRange> {
let mut missing = Vec::new();
for (author, peer_info) in peer.authors() {
let my_seq = self.seq(author);
let my_heads = self.heads(author);
// We need entries if:
// 1. Peer's seq is higher than ours, OR
// 2. Peer's seq equals ours but they have heads we don't (fork)
let need_entries = if peer_info.seq > my_seq {
true
} else if peer_info.seq == my_seq && my_seq > 0 {
// Same seq - check for forks (different hashes at same seq)
peer_info.heads.iter().any(|h| !my_heads.contains(h))
} else {
false
};
if need_entries {
// Request from our common ancestor (or start if we have nothing)
let from_hash = if my_heads.is_empty() {
[0u8; 32]
} else {
*my_heads.iter().next().unwrap()
};
missing.push(MissingRange {
author: *author,
from_seq: my_seq,
from_hash,
to_seq: peer_info.seq,
});
}
}
missing
}
/// Merge another sync state into this one (union of heads, max seq).
pub fn merge(&mut self, other: &SyncState) {
for (author, info) in other.authors() {
if let Some(my_info) = self.authors.get_mut(author) {
// Union heads
for h in &info.heads {
my_info.heads.insert(*h);
}
// Take max seq
if info.seq > my_info.seq {
my_info.seq = info.seq;
}
} else {
self.authors.insert(*author, info.clone());
}
}
}
/// Convert to proto message for network transmission
pub fn to_proto(&self) -> crate::proto::SyncState {
let frontiers = self.authors.iter().map(|(author, info)| {
crate::proto::Frontier {
author_id: author.to_vec(),
max_seq: info.seq,
head_hashes: info.heads.iter().map(|h| h.to_vec()).collect(),
}
}).collect();
crate::proto::SyncState {
frontiers,
sender_hlc: None,
}
}
/// Create from proto message
pub fn from_proto(proto: &crate::proto::SyncState) -> Self {
let mut state = Self::new();
for frontier in &proto.frontiers {
if frontier.author_id.len() == 32 {
let mut author = [0u8; 32];
author.copy_from_slice(&frontier.author_id);
let mut heads = HashSet::new();
for hash_bytes in &frontier.head_hashes {
if hash_bytes.len() == 32 {
let mut hash = [0u8; 32];
hash.copy_from_slice(hash_bytes);
heads.insert(hash);
}
}
if heads.is_empty() {
// Fallback: empty hash if no heads provided
heads.insert([0u8; 32]);
}
state.set_heads(author, frontier.max_seq, heads);
}
}
state
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_diff_empty() {
let a = SyncState::new();
let b = SyncState::new();
assert!(a.diff(&b).is_empty());
}
#[test]
fn test_diff_peer_ahead() {
let mut a = SyncState::new();
let mut b = SyncState::new();
let author = [1u8; 32];
let hash_a = [0xAA; 32];
let hash_b = [0xBB; 32];
a.set(author, 5, hash_a);
b.set(author, 10, hash_b);
let missing = a.diff(&b);
assert_eq!(missing.len(), 1);
assert_eq!(missing[0].author, author);
assert_eq!(missing[0].from_seq, 5);
assert_eq!(missing[0].from_hash, hash_a); // Resume after our hash
assert_eq!(missing[0].to_seq, 10);
}
#[test]
fn test_diff_i_am_ahead() {
let mut a = SyncState::new();
let mut b = SyncState::new();
let author = [1u8; 32];
a.set(author, 10, [0xAA; 32]);
b.set(author, 5, [0xBB; 32]);
// I'm ahead, so I don't need anything from peer
let missing = a.diff(&b);
assert!(missing.is_empty());
}
#[test]
fn test_diff_new_author() {
let a = SyncState::new();
let mut b = SyncState::new();
let author = [2u8; 32];
b.set(author, 3, [0xBB; 32]);
// Peer has author I don't have
let missing = a.diff(&b);
assert_eq!(missing.len(), 1);
assert_eq!(missing[0].from_seq, 0);
assert_eq!(missing[0].from_hash, [0u8; 32]); // Zero hash = read from start
assert_eq!(missing[0].to_seq, 3);
}
#[test]
fn test_merge() {
let mut a = SyncState::new();
let mut b = SyncState::new();
let author1 = [1u8; 32];
let author2 = [2u8; 32];
a.set(author1, 10, [0xA1; 32]);
a.set(author2, 5, [0xA2; 32]);
b.set(author1, 5, [0xB1; 32]); // a is ahead
b.set(author2, 8, [0xB2; 32]); // b is ahead
a.merge(&b);
assert_eq!(a.seq(&author1), 10); // kept a's value
assert_eq!(a.seq(&author2), 8); // took b's value
}
/// This test documents a known issue: SyncState tracks only ONE hash per author,
/// but with forks/multi-heads, there could be multiple branches.
///
/// Scenario:
/// - Author writes entry1 (hash=A)
/// - Two peers independently write entry2 and entry3 (both have prev=A)
/// - Peer1 has: entry1 -> entry2 (seq=2, hash=B)
/// - Peer2 has: entry1 -> entry3 (seq=2, hash=C)
/// - When Peer3 syncs with Peer1, SyncState says "I need entries after hash=B"
/// - But Peer2 only has entries after hash=A, so Peer3 never gets entry3!
///
#[test]
fn test_multihead_sync_inconsistency() {
// This is a conceptual test showing the problem
// In reality, both forks would have seq=2 but different hashes
// SyncState can only track one, so the other branch gets lost
let mut peer1_state = SyncState::new();
let mut peer2_state = SyncState::new();
let new_peer_state = SyncState::new();
let author = [1u8; 32];
// Both peers have seq=2, but different hashes (different forks)
peer1_state.set(author, 2, [0xBB; 32]); // entry1 -> entry2
peer2_state.set(author, 2, [0xCC; 32]); // entry1 -> entry3
// New peer syncs with peer1 first
let missing_from_peer1 = new_peer_state.diff(&peer1_state);
assert_eq!(missing_from_peer1.len(), 1);
assert_eq!(missing_from_peer1[0].to_seq, 2);
// After applying peer1's entries, new peer has seq=2, hash=BB
let mut after_peer1 = new_peer_state.clone();
after_peer1.set(author, 2, [0xBB; 32]);
// Now sync with peer2 - BUG: new peer thinks it's up to date!
let missing_from_peer2 = after_peer1.diff(&peer2_state);
// This assertion FAILS - we get empty missing even though peer2 has entry3!
// The bug: peer2's seq=2 equals our seq=2, so we think we're in sync
// But peer2's hash=0xCC != our hash=0xBB - they have different entries!
assert!(!missing_from_peer2.is_empty(),
"BUG: SyncState misses peer2's fork because seq numbers match");
}
}
-41
View File
@@ -1,41 +0,0 @@
//! Vector clocks for causality tracking
use std::collections::HashMap;
/// A vector clock for tracking "how much" of each node's log has been seen.
///
/// Used during reconciliation to identify missing entries between peers.
pub struct VectorClock {
clocks: HashMap<[u8; 32], u64>,
}
impl VectorClock {
/// Create a new empty vector clock.
pub fn new() -> Self {
Self {
clocks: HashMap::new(),
}
}
/// Get the clock value for a node (returns 0 if not present).
pub fn get(&self, node_id: &[u8; 32]) -> u64 {
self.clocks.get(node_id).copied().unwrap_or(0)
}
/// Set the clock value for a node.
pub fn set(&mut self, node_id: [u8; 32], value: u64) {
self.clocks.insert(node_id, value);
}
/// Increment the clock for a node.
pub fn increment(&mut self, node_id: [u8; 32]) {
let current = self.get(&node_id);
self.set(node_id, current + 1);
}
}
impl Default for VectorClock {
fn default() -> Self {
Self::new()
}
}
+7
View File
@@ -9,10 +9,17 @@ license.workspace = true
lattice-core = { workspace = true }
iroh = { workspace = true }
iroh-gossip = { workspace = true }
prost = { workspace = true }
tokio = { workspace = true }
thiserror = { workspace = true }
tracing = { workspace = true }
bytes = { workspace = true }
tokio-util = { workspace = true }
futures-util = { workspace = true }
hex = { workspace = true }
blake3.workspace = true
anyhow = "1.0.100"
futures-lite = "2.6.1"
[dev-dependencies]
tokio-test = { workspace = true }
+60
View File
@@ -0,0 +1,60 @@
//! Iroh endpoint for network connectivity
//!
//! Creates an Iroh endpoint from the node's Ed25519 secret key,
//! ensuring the same identity is used for both Lattice and Iroh.
//!
//! Discovery: Uses both DNS (default) and mDNS (local network)
use iroh::{Endpoint, endpoint::{BindError, Connection, ConnectError}};
use iroh::discovery::mdns::MdnsDiscovery;
pub use iroh::PublicKey;
/// ALPN protocol identifier for Lattice sync
pub const LATTICE_ALPN: &[u8] = b"lattice-sync/1";
/// Wrapper around Iroh endpoint with Lattice integration
pub struct LatticeEndpoint {
endpoint: Endpoint,
}
impl LatticeEndpoint {
/// Create a new endpoint from Ed25519 secret key bytes (from identity.key)
/// Enables both DNS discovery (internet) and mDNS discovery (local network)
pub async fn new(secret_key_bytes: [u8; 32]) -> Result<Self, BindError> {
let secret_key = iroh::SecretKey::from_bytes(&secret_key_bytes);
// mDNS for local network discovery
let mdns = MdnsDiscovery::builder();
let endpoint = Endpoint::builder()
.secret_key(secret_key)
.alpns(vec![
LATTICE_ALPN.to_vec(),
iroh_gossip::ALPN.to_vec(), // Also accept gossip protocol
])
.discovery(mdns) // Add mDNS on top of default DNS
.bind()
.await?;
Ok(Self { endpoint })
}
/// Get the public key (same as Lattice pubkey, can be shared with peers)
pub fn public_key(&self) -> PublicKey {
self.endpoint.secret_key().public()
}
/// Connect to a peer by their public key
pub async fn connect(&self, peer: PublicKey) -> Result<Connection, ConnectError> {
self.endpoint.connect(peer, LATTICE_ALPN).await
}
/// Accept an incoming connection
pub async fn accept(&self) -> Option<iroh::endpoint::Incoming> {
self.endpoint.accept().await
}
/// Get the underlying endpoint
pub fn endpoint(&self) -> &Endpoint {
&self.endpoint
}
}
+63
View File
@@ -0,0 +1,63 @@
//! Message framing for Iroh streams using tokio-util LengthDelimitedCodec
//!
//! Provides a clean interface for sending/receiving length-prefixed PeerMessage
//! over QUIC streams without manual buffer management.
use futures_util::{SinkExt, StreamExt};
use lattice_core::proto::PeerMessage;
use prost::Message;
use tokio_util::codec::{FramedRead, FramedWrite, LengthDelimitedCodec};
/// Framed writer for sending PeerMessage over an Iroh SendStream
pub struct MessageSink {
inner: FramedWrite<iroh::endpoint::SendStream, LengthDelimitedCodec>,
}
impl MessageSink {
pub fn new(stream: iroh::endpoint::SendStream) -> Self {
Self {
inner: FramedWrite::new(stream, LengthDelimitedCodec::new()),
}
}
/// Send a PeerMessage (length-prefixed)
pub async fn send(&mut self, msg: &PeerMessage) -> Result<(), String> {
let bytes = msg.encode_to_vec();
self.inner.send(bytes.into()).await
.map_err(|e| format!("Send error: {}", e))
}
/// Finish the stream (signal we're done sending)
pub async fn finish(self) -> Result<(), String> {
let mut stream = self.inner.into_inner();
let _ = stream.finish();
stream.stopped().await.ok();
Ok(())
}
}
/// Framed reader for receiving PeerMessage from an Iroh RecvStream
pub struct MessageStream {
inner: FramedRead<iroh::endpoint::RecvStream, LengthDelimitedCodec>,
}
impl MessageStream {
pub fn new(stream: iroh::endpoint::RecvStream) -> Self {
Self {
inner: FramedRead::new(stream, LengthDelimitedCodec::new()),
}
}
/// Receive next PeerMessage (or None if stream closed)
pub async fn recv(&mut self) -> Result<Option<PeerMessage>, String> {
match self.inner.next().await {
Some(Ok(bytes)) => {
PeerMessage::decode(&bytes[..])
.map(Some)
.map_err(|e| format!("Decode error: {}", e))
}
Some(Err(e)) => Err(format!("Read error: {}", e)),
None => Ok(None),
}
}
}
+16 -1
View File
@@ -1,8 +1,23 @@
//! Lattice Networking
//!
//! Networking layer using Iroh:
//! - **Endpoint**: Network identity and connection management
//! - **Gossip**: Broadcasting changes across the mesh
//! - **Unicast**: Point-to-point communication for reconciliation
//! - **Framing**: Length-delimited message framing for QUIC streams
//! - **Mesh**: Peer-to-peer join and sync operations
pub mod endpoint;
pub mod gossip;
pub mod unicast;
pub mod framing;
pub mod mesh;
pub use endpoint::{LatticeEndpoint, PublicKey, LATTICE_ALPN};
pub use framing::{MessageSink, MessageStream};
pub use lattice_core::proto::{SyncRequest, SyncResponse, SyncEntry, SyncDone, SyncState, Frontier};
pub use mesh::{LatticeServer, SyncResult};
/// Parse a PublicKey (NodeId) from hex or base32 string
pub fn parse_node_id(s: &str) -> Result<PublicKey, String> {
s.parse().map_err(|e| format!("{}", e))
}
+10
View File
@@ -0,0 +1,10 @@
//! Mesh networking - peer-to-peer join and sync operations
//!
//! - **server**: LatticeServer for mesh networking (join, sync, accept loop)
//! - **protocol**: Shared send/receive entry logic
mod server;
mod protocol;
pub use server::{LatticeServer, SyncResult};
pub use protocol::{send_missing_entries, receive_entries};
+86
View File
@@ -0,0 +1,86 @@
//! Protocol - shared logic for bidirectional sync entry exchange
use crate::{MessageSink, MessageStream};
use lattice_core::{StoreHandle, CausalEntryIter};
use lattice_core::proto::{peer_message, PeerMessage, SignedEntry};
use lattice_core::sync_state::SyncState;
use prost::Message;
use std::collections::VecDeque;
/// Send entries that peer is missing based on state diff.
/// Returns (entries_sent, optional_error).
pub async fn send_missing_entries(
sink: &mut MessageSink,
store: &StoreHandle,
my_state: &SyncState,
peer_state: &SyncState,
) -> Result<u64, String> {
let missing = peer_state.diff(my_state);
// Build queues for each author's entries
let mut author_entries: Vec<VecDeque<SignedEntry>> = Vec::new();
for range in missing {
let from_hash = if range.from_hash == [0u8; 32] { None } else { Some(range.from_hash) };
let entries = store.read_entries_after(&range.author, from_hash).await
.map_err(|e| format!("Failed to read entries: {}", e))?;
if !entries.is_empty() {
author_entries.push(entries.into());
}
}
// Stream entries in HLC (causal) order
let mut entries_sent = 0u64;
for entry in CausalEntryIter::new(author_entries) {
let sync_msg = PeerMessage {
message: Some(peer_message::Message::SyncEntry(lattice_core::proto::SyncEntry {
signed_entry: entry.encode_to_vec(),
hash: vec![],
})),
};
sink.send(&sync_msg).await?;
entries_sent += 1;
}
// Send SyncDone
let done = PeerMessage {
message: Some(peer_message::Message::SyncDone(lattice_core::proto::SyncDone {
entries_sent,
})),
};
sink.send(&done).await?;
Ok(entries_sent)
}
/// Receive and apply entries until SyncDone is received.
/// Returns (entries_applied, entries_reported_by_peer).
pub async fn receive_entries(
stream: &mut MessageStream,
store: &StoreHandle,
) -> Result<(u64, u64), String> {
let mut entries_applied = 0u64;
let mut entries_reported = 0u64;
loop {
match stream.recv().await {
Ok(Some(msg)) => match msg.message {
Some(peer_message::Message::SyncEntry(entry)) => {
if let Ok(signed) = SignedEntry::decode(&entry.signed_entry[..]) {
if store.apply_entry(signed).await.is_ok() {
entries_applied += 1;
}
}
}
Some(peer_message::Message::SyncDone(done)) => {
entries_reported = done.entries_sent;
break;
}
_ => {}
}
Ok(None) => break,
Err(_) => break,
}
}
Ok((entries_applied, entries_reported))
}
+575
View File
@@ -0,0 +1,575 @@
//! Server - LatticeServer for mesh networking
use crate::{MessageSink, MessageStream, LatticeEndpoint, parse_node_id, LATTICE_ALPN};
use lattice_core::{Node, NodeError, NodeEvent, PeerStatus, Uuid, StoreHandle};
use iroh::endpoint::Connection;
use iroh::protocol::{Router, ProtocolHandler, AcceptError};
use iroh_gossip::Gossip;
use std::sync::Arc;
use std::collections::HashMap;
use tokio::sync::RwLock;
use futures_util::StreamExt;
use lattice_core::proto::{PeerMessage, peer_message, JoinRequest, JoinResponse, SignedEntry};
use prost::Message;
use super::protocol;
/// Result of a sync operation with a peer
pub struct SyncResult {
pub entries_applied: u64,
pub entries_sent_by_peer: u64,
}
/// LatticeServer wraps Node + Endpoint + Gossip and provides mesh networking methods.
/// Uses Router to handle incoming connections for both sync and gossip protocols.
pub struct LatticeServer {
node: Arc<Node>,
endpoint: LatticeEndpoint,
gossip: Gossip,
#[allow(dead_code)]
router: Router,
/// Gossip senders per store topic
gossip_senders: Arc<RwLock<HashMap<Uuid, iroh_gossip::api::GossipSender>>>,
}
/// Protocol handler for lattice sync connections
struct SyncProtocol {
node: Arc<Node>,
}
impl std::fmt::Debug for SyncProtocol {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SyncProtocol").finish()
}
}
impl ProtocolHandler for SyncProtocol {
fn accept(&self, conn: Connection) -> impl std::future::Future<Output = Result<(), AcceptError>> + Send {
let node = self.node.clone();
Box::pin(async move {
if let Err(e) = handle_connection(node, conn).await {
eprintln!("[Accept] Error: {}", e);
// Log error but return Ok - protocol handled the connection
}
Ok(())
})
}
}
impl LatticeServer {
/// Create a new LatticeServer from just a Node (creates endpoint internally).
pub async fn new_from_node(node: Arc<Node>) -> Result<Self, String> {
let endpoint = LatticeEndpoint::new(node.secret_key_bytes()).await
.map_err(|e| format!("Failed to create endpoint: {}", e))?;
Self::new(node, endpoint).await
}
/// Create a new LatticeServer with existing endpoint.
pub async fn new(node: Arc<Node>, endpoint: LatticeEndpoint) -> Result<Self, String> {
// Create gossip instance
let gossip = Gossip::builder().spawn(endpoint.endpoint().clone());
// Create sync protocol handler
let sync_protocol = SyncProtocol { node: node.clone() };
// Create router to handle both protocols
let router = Router::builder(endpoint.endpoint().clone())
.accept(LATTICE_ALPN, sync_protocol)
.accept(iroh_gossip::ALPN, gossip.clone())
.spawn();
let server = Self {
node,
endpoint,
gossip,
router,
gossip_senders: Arc::new(RwLock::new(HashMap::new())),
};
server.spawn_node_event_listener();
// If root store is already open, start gossip for it
if let Some(store) = (*server.node.root_store().await).clone() {
println!("[Gossip] Root store already open, starting gossip...");
server.join_gossip_topic(store.id()).await?;
server.spawn_entry_forward_loop(store);
}
Ok(server)
}
/// Spawn a listener for Node events (auto-starts gossip when root store is activated)
fn spawn_node_event_listener(&self) {
let mut event_rx = self.node.subscribe_events();
let gossip_senders = self.gossip_senders.clone();
let gossip = self.gossip.clone();
let node = self.node.clone();
tokio::spawn(async move {
while let Ok(event) = event_rx.recv().await {
match event {
NodeEvent::RootStoreActivated(store) => {
println!("[Gossip] Root store activated: {}, starting gossip...", store.id());
let store_id = store.id();
// Get bootstrap peers from node's peer list
let bootstrap_peers: Vec<iroh::PublicKey> = match node.list_peers().await {
Ok(peers) => {
peers.iter()
.filter(|p| p.status == PeerStatus::Active)
.filter_map(|p| parse_node_id(&p.pubkey).ok())
.collect()
}
Err(e) => {
eprintln!("[Gossip] Failed to list peers: {}, using empty list", e);
Vec::new()
}
};
println!("[Gossip] Bootstrap peers: {}", bootstrap_peers.len());
// Topic ID from hash of "lattice/{store_id}" for namespacing
let topic_bytes = blake3::hash(format!("lattice/{}", store_id).as_bytes());
let topic_id = iroh_gossip::TopicId::from_bytes(*topic_bytes.as_bytes());
// Use subscribe (non-blocking) - peers will connect when they sync
// subscribe_and_join would block waiting for peers we can't reach yet
match gossip.subscribe(topic_id, bootstrap_peers).await {
Ok(sub) => {
let (sender, receiver) = sub.split();
// Store sender
gossip_senders.write().await.insert(store_id, sender);
// Spawn receive loop
let store_recv = store.clone();
tokio::spawn(async move {
let mut receiver = receiver;
println!("[Gossip] Receive loop started for topic {:?}", topic_id);
while let Some(event) = futures_util::StreamExt::next(&mut receiver).await {
match event {
Ok(iroh_gossip::api::Event::Received(msg)) => {
println!("[Gossip] Received {} bytes", msg.content.len());
if let Ok(entry) = SignedEntry::decode(&msg.content[..]) {
if let Err(e) = store_recv.apply_entry(entry).await {
eprintln!("[Gossip] Failed to apply entry: {}", e);
} else {
println!("[Gossip] Applied entry successfully");
}
}
}
Ok(other) => {
println!("[Gossip] Event: {:?}", other);
}
Err(e) => {
eprintln!("[Gossip] Error: {}", e);
}
}
}
});
// Spawn entry forward loop
let gossip_senders = gossip_senders.clone();
let mut entry_rx = store.subscribe_entries();
tokio::spawn(async move {
println!("[Gossip] Entry forward loop started for store {}", store_id);
while let Ok(entry) = entry_rx.recv().await {
let senders = gossip_senders.read().await;
if let Some(sender) = senders.get(&store_id) {
let bytes = entry.encode_to_vec();
println!("[Gossip] Broadcasting {} bytes", bytes.len());
let _ = sender.broadcast(bytes.into()).await;
}
}
});
println!("[Gossip] Gossip started for store {}", store_id);
}
Err(e) => {
eprintln!("[Gossip] Failed to subscribe to topic: {}", e);
}
}
}
}
}
});
}
/// Start gossip for a store (call after store is opened)
pub async fn start_gossip_for_store(&self, store: StoreHandle) -> Result<(), String> {
println!("[Gossip] Starting gossip for store {}", store.id());
self.join_gossip_topic(store.id()).await?;
self.spawn_entry_forward_loop(store);
Ok(())
}
/// Spawn a loop that forwards local store entry broadcasts to gossip
fn spawn_entry_forward_loop(&self, store: StoreHandle) {
let store_id = store.id();
let gossip_senders = self.gossip_senders.clone();
let mut entry_rx = store.subscribe_entries();
println!("[Gossip] Starting entry forward loop for store {}", store_id);
tokio::spawn(async move {
while let Ok(entry) = entry_rx.recv().await {
println!("[Gossip] Received local entry, forwarding to gossip...");
// Forward to gossip sender
let senders = gossip_senders.read().await;
if let Some(sender) = senders.get(&store_id) {
let bytes = entry.encode_to_vec();
println!("[Gossip] Broadcasting {} bytes to topic {}", bytes.len(), store_id);
if let Err(e) = sender.broadcast(bytes.into()).await {
eprintln!("[Gossip] Failed to broadcast entry: {}", e);
} else {
println!("[Gossip] Broadcast successful");
}
} else {
eprintln!("[Gossip] No gossip sender for store {}", store_id);
}
}
println!("[Gossip] Entry forward loop ended for store {}", store_id);
});
}
/// Access the underlying node
pub fn node(&self) -> &Node {
&self.node
}
/// Access the underlying endpoint
pub fn endpoint(&self) -> &LatticeEndpoint {
&self.endpoint
}
/// Join gossip topic for a store (subscribes and spawns receive loop)
pub async fn join_gossip_topic(&self, store_id: Uuid) -> Result<(), String> {
// Get active peers to bootstrap gossip
let peers = self.node.list_peers().await
.map_err(|e| format!("Failed to list peers: {}", e))?;
let bootstrap_peers: Vec<iroh::PublicKey> = peers.iter()
.filter(|p| p.status == PeerStatus::Active)
.filter_map(|p| parse_node_id(&p.pubkey).ok())
.collect();
println!("[Gossip] Joining topic {} with {} bootstrap peers", store_id, bootstrap_peers.len());
// Topic ID from store UUID bytes (padded to 32 bytes)
// Topic ID from hash of "lattice/{store_id}" for namespacing
let topic_bytes = blake3::hash(format!("lattice/{}", store_id).as_bytes());
let topic_id = iroh_gossip::TopicId::from_bytes(*topic_bytes.as_bytes());
// Subscribe to topic
let (sender, mut receiver) = self.gossip.subscribe(topic_id, bootstrap_peers).await
.map_err(|e| format!("Failed to subscribe to gossip topic: {}", e))?
.split();
// Store sender for broadcasting
{
let mut senders = self.gossip_senders.write().await;
senders.insert(store_id, sender);
}
// Spawn receive loop
let node = self.node.clone();
let topic = topic_id;
tokio::spawn(async move {
// StreamExt imported at module level
println!("[Gossip] Receive loop started for topic {:?}", topic);
while let Some(event) = receiver.next().await {
match event {
Ok(iroh_gossip::api::Event::Received(message)) => {
println!("[Gossip] Received gossip message: {} bytes", message.content.len());
// Decode SignedEntry and apply
match SignedEntry::decode(&message.content[..]) {
Ok(entry) => {
println!("[Gossip] Decoded entry, applying...");
// Find store and apply entry
if let Some(store) = (*node.root_store().await).clone() {
if let Err(e) = store.apply_entry(entry.into()).await {
eprintln!("[Gossip] Failed to apply entry: {}", e);
} else {
println!("[Gossip] Entry applied successfully");
}
} else {
eprintln!("[Gossip] No root store to apply entry to");
}
}
Err(e) => eprintln!("[Gossip] Failed to decode entry: {}", e),
}
}
Ok(other) => {
println!("[Gossip] Other event: {:?}", other);
}
Err(e) => eprintln!("[Gossip] Receive error: {}", e),
}
}
println!("[Gossip] Receive loop ended for topic");
});
Ok(())
}
/// Broadcast an entry to all gossip subscribers for a store
pub async fn broadcast_entry(&self, store_id: Uuid, entry: &SignedEntry) -> Result<(), String> {
let senders = self.gossip_senders.read().await;
if let Some(sender) = senders.get(&store_id) {
let bytes = entry.encode_to_vec();
sender.broadcast(bytes.into()).await
.map_err(|e| format!("Gossip broadcast failed: {}", e))?;
}
Ok(())
}
/// Join an existing mesh by connecting to a peer.
pub async fn join_mesh(&self, peer_id: iroh::PublicKey) -> Result<StoreHandle, NodeError> {
let conn = self.endpoint.connect(peer_id).await
.map_err(|e| NodeError::Actor(format!("Connection failed: {}", e)))?;
let (send, recv) = conn.open_bi().await
.map_err(|e| NodeError::Actor(format!("Failed to open stream: {}", e)))?;
let mut sink = MessageSink::new(send);
let mut stream = MessageStream::new(recv);
// Send JoinRequest
let req = PeerMessage {
message: Some(peer_message::Message::JoinRequest(JoinRequest {
node_pubkey: self.node.node_id().to_vec(),
})),
};
sink.send(&req).await.map_err(|e| NodeError::Actor(e))?;
sink.finish().await.map_err(|e| NodeError::Actor(e))?;
// Receive JoinResponse
let msg = stream.recv().await
.map_err(|e| NodeError::Actor(e))?
.ok_or_else(|| NodeError::Actor("Peer closed stream".to_string()))?;
match msg.message {
Some(peer_message::Message::JoinResponse(resp)) => {
let store_uuid = lattice_core::Uuid::from_slice(&resp.store_uuid)
.map_err(|_| NodeError::Actor("Invalid UUID from peer".to_string()))?;
let handle = self.node.complete_join(store_uuid).await?;
// Sync with peer to get initial data
println!("[Join] Syncing with peer to get initial data...");
if let Ok(result) = self.sync_with_peer(&handle, peer_id).await {
println!("[Join] Initial sync complete: {} entries", result.entries_applied);
}
Ok(handle)
}
_ => Err(NodeError::Actor("Unexpected response".to_string())),
}
}
/// Sync with a specific peer.
pub async fn sync_with_peer(&self, store: &StoreHandle, peer_id: iroh::PublicKey) -> Result<SyncResult, NodeError> {
let conn = self.endpoint.connect(peer_id).await
.map_err(|e| NodeError::Actor(format!("Connection failed: {}", e)))?;
let (send, recv) = conn.open_bi().await
.map_err(|e| NodeError::Actor(format!("Failed to open stream: {}", e)))?;
let mut sink = MessageSink::new(send);
let mut stream = MessageStream::new(recv);
let my_state = store.sync_state().await?;
// Send SyncRequest
let req = PeerMessage {
message: Some(peer_message::Message::SyncRequest(lattice_core::proto::SyncRequest {
store_id: store.id().as_bytes().to_vec(),
state: Some(my_state.to_proto()),
full_sync: false,
})),
};
sink.send(&req).await.map_err(|e| NodeError::Actor(e))?;
// Receive SyncResponse
let resp_msg = stream.recv().await.map_err(|e| NodeError::Actor(e))?
.ok_or_else(|| NodeError::Actor("Peer closed stream".to_string()))?;
let peer_state = match resp_msg.message {
Some(peer_message::Message::SyncResponse(resp)) => {
resp.state.map(|s| lattice_core::sync_state::SyncState::from_proto(&s))
.unwrap_or_default()
}
_ => return Err(NodeError::Actor("Expected SyncResponse".to_string())),
};
// Exchange entries
let _entries_sent = protocol::send_missing_entries(&mut sink, store, &my_state, &peer_state).await
.map_err(|e| NodeError::Actor(e))?;
let (entries_applied, entries_sent_by_peer) = protocol::receive_entries(&mut stream, store).await
.map_err(|e| NodeError::Actor(e))?;
sink.finish().await.map_err(|e| NodeError::Actor(e))?;
Ok(SyncResult { entries_applied, entries_sent_by_peer })
}
/// Sync with all active peers.
pub async fn sync_all(&self, store: &StoreHandle) -> Result<Vec<SyncResult>, NodeError> {
let peers = self.node.list_peers().await?;
let mut results = Vec::new();
let my_pubkey = self.endpoint.public_key();
for peer in peers {
if peer.status != PeerStatus::Active {
continue;
}
let peer_id = match parse_node_id(&peer.pubkey) {
Ok(id) => id,
Err(e) => {
eprintln!("[Sync] Failed to parse peer {}: {}", peer.pubkey, e);
continue;
}
};
// Skip self
if peer_id == my_pubkey {
continue;
}
println!("[Sync] Syncing with {}...", peer_id.fmt_short());
match self.sync_with_peer(store, peer_id).await {
Ok(result) => {
println!("[Sync] Applied {} entries", result.entries_applied);
results.push(result);
}
Err(e) => eprintln!("[Sync] Failed: {}", e),
}
}
Ok(results)
}
}
// --- Connection handling ---
// --- Connection handling ---
/// Handle a single incoming connection
async fn handle_connection(
node: Arc<Node>,
conn: Connection,
) -> Result<(), String> {
let remote_id = conn.remote_id();
let remote_hex = hex::encode(remote_id.as_bytes());
println!("\n[Incoming] {} (ALPN: {})", remote_id.fmt_short(), String::from_utf8_lossy(conn.alpn()));
// Parse remote pubkey
let remote_pubkey: [u8; 32] = hex::decode(&remote_hex)
.map_err(|_| "Invalid pubkey hex")?
.try_into()
.map_err(|_| "Invalid pubkey length")?;
let (send, recv) = conn.accept_bi().await
.map_err(|e| format!("Accept stream error: {}", e))?;
let sink = MessageSink::new(send);
let mut stream = MessageStream::new(recv);
// Read first message to determine request type
let msg = stream.recv().await?
.ok_or_else(|| "Peer closed stream".to_string())?;
match msg.message {
Some(peer_message::Message::JoinRequest(req)) => {
handle_join_request(&node, &remote_pubkey, req, sink).await
}
Some(peer_message::Message::SyncRequest(req)) => {
handle_sync_request(&node, &remote_pubkey, req, sink, stream).await
}
_ => Err("Unexpected message type".to_string()),
}
}
/// Handle a join request from an invited peer
async fn handle_join_request(
node: &Node,
remote_pubkey: &[u8; 32],
req: lattice_core::proto::JoinRequest,
mut sink: MessageSink,
) -> Result<(), String> {
println!("[Join] Got JoinRequest from {}", hex::encode(&req.node_pubkey));
// Accept the join - verifies invited, sets active, returns store ID
let acceptance = node.accept_join(remote_pubkey).await
.map_err(|e| e.to_string())?;
let resp = PeerMessage {
message: Some(peer_message::Message::JoinResponse(JoinResponse {
store_uuid: acceptance.store_id.as_bytes().to_vec(),
inviter_pubkey: vec![],
})),
};
sink.send(&resp).await?;
sink.finish().await?;
println!("[Join] Sent JoinResponse, peer now active");
Ok(())
}
/// Handle a sync request - bidirectional exchange of entries
async fn handle_sync_request(
node: &Node,
remote_pubkey: &[u8; 32],
peer_request: lattice_core::proto::SyncRequest,
mut sink: MessageSink,
mut stream: MessageStream,
) -> Result<(), String> {
// Verify peer is active (allowed to sync)
node.verify_peer_status(remote_pubkey, &[PeerStatus::Active]).await
.map_err(|e| e.to_string())?;
println!("[Sync] Verified peer as active");
// Parse store_id from request
let store_id = Uuid::from_slice(&peer_request.store_id)
.map_err(|_| format!("Invalid store_id in SyncRequest: {} bytes, expected 16", peer_request.store_id.len()))?;
println!("[Sync] Received SyncRequest for store {}", store_id);
// Open the requested store (uses cache if available)
let (store, _info) = node.open_store(store_id).await
.map_err(|e| format!("Failed to open store {}: {}", store_id, e))?;
println!("[Sync] Received SyncRequest");
// Get our sync state
let my_state = store.sync_state().await
.map_err(|e| format!("Failed to get sync state: {}", e))?;
// 1. Send our sync state as response
let resp = PeerMessage {
message: Some(peer_message::Message::SyncResponse(lattice_core::proto::SyncResponse {
store_id: store.id().as_bytes().to_vec(),
state: Some(my_state.to_proto()),
})),
};
sink.send(&resp).await?;
// 2. Send entries peer is missing
let peer_state = peer_request.state
.map(|s| lattice_core::sync_state::SyncState::from_proto(&s))
.unwrap_or_default();
let entries_sent = protocol::send_missing_entries(&mut sink, &store, &my_state, &peer_state).await?;
println!("[Sync] Sent {} entries, now receiving from peer...", entries_sent);
// 3. Receive entries from requester (bidirectional)
let (entries_applied, _) = protocol::receive_entries(&mut stream, &store).await?;
sink.finish().await?;
println!("[Sync] Applied {} entries from peer", entries_applied);
Ok(())
}
-3
View File
@@ -1,3 +0,0 @@
//! Unicast communication for direct peer-to-peer messaging
// TODO: Implement unicast using iroh
-15
View File
@@ -1,15 +0,0 @@
[package]
name = "lattice-store"
description = "Log-based KV store with snapshots and watermarks"
version.workspace = true
edition.workspace = true
license.workspace = true
[dependencies]
lattice-core = { workspace = true }
tokio = { workspace = true }
thiserror = { workspace = true }
tracing = { workspace = true }
[dev-dependencies]
tokio-test = { workspace = true }
-12
View File
@@ -1,12 +0,0 @@
//! Lattice Store
//!
//! Log-based Key-Value store:
//! - State is a "view" generated by replaying entries
//! - Watermarks for agreeing on safe compaction points
//! - Snapshots for replacing old logs
pub mod store;
pub mod snapshot;
pub mod watermark;
pub use store::Store;
-6
View File
@@ -1,6 +0,0 @@
//! Snapshots for log compaction
/// A verified snapshot that can replace old log entries.
pub struct Snapshot {
// TODO: snapshot data, watermark, verification
}
-35
View File
@@ -1,35 +0,0 @@
//! The main KV store
use std::collections::HashMap;
/// A log-based Key-Value store.
///
/// The store is a dynamic "view" generated by replaying entries.
pub struct Store {
data: HashMap<Vec<u8>, Vec<u8>>,
}
impl Store {
/// Create a new empty store.
pub fn new() -> Self {
Self {
data: HashMap::new(),
}
}
/// Get a value by key.
pub fn get(&self, key: &[u8]) -> Option<&[u8]> {
self.data.get(key).map(|v| v.as_slice())
}
/// Set a value (this would normally go through the log).
pub fn set(&mut self, key: Vec<u8>, value: Vec<u8>) {
self.data.insert(key, value);
}
}
impl Default for Store {
fn default() -> Self {
Self::new()
}
}
-9
View File
@@ -1,9 +0,0 @@
//! Watermarks for agreeing on safe compaction points
/// A watermark representing a safe cut-off point for log compaction.
///
/// Nodes use watermarks to agree on which entries can be safely
/// deleted and replaced with snapshots.
pub struct Watermark {
// TODO: watermark data
}
+75 -5
View File
@@ -20,15 +20,42 @@ message Entry {
// Versioning allows us to change the format radically later if needed
uint32 version = 1;
// Store this entry belongs to (16-byte UUID)
bytes store_id = 6;
// Ordering Metadata
bytes prev_hash = 2; // Link to previous entry (32 bytes)
bytes prev_hash = 2; // Link to previous sigchain entry (32 bytes)
uint64 seq = 3; // Monotonic sequence number
HLC timestamp = 4; // Hybrid Logical Clock
// DAG ancestry: hashes of entries this supersedes (separate from sigchain)
repeated bytes parent_hashes = 7;
// The Batch of Operations
repeated Operation ops = 5;
}
// HeadInfo: a tip/head in the DAG for a key
message HeadInfo {
bytes value = 1; // The value at this head
uint64 hlc = 2; // Combined HLC for ordering (wall_time_ms << 16 | counter)
bytes author = 3; // Author's public key (32 bytes)
bytes hash = 4; // Hash of the SignedEntry that created this head
bool tombstone = 5; // True if this head represents a delete
}
// HeadList: wrapper for storing multiple heads per key in state.db
message HeadList {
repeated HeadInfo heads = 1;
}
// AuthorState: tracks last applied entry per author for replay optimization
message AuthorState {
uint64 seq = 1; // Last applied seq for this author's sigchain
bytes hash = 2; // Hash of last applied entry
uint64 log_offset = 3; // Byte offset in log file for fast resume
}
// Hybrid Logical Clock
message HLC {
uint64 wall_time = 1; // Unix timestamp (ms)
@@ -46,12 +73,12 @@ message Operation {
}
message PutOp {
string key = 1;
bytes value = 2; // Raw bytes allows storing images, JSON, binary, etc.
bytes key = 1;
bytes value = 2;
}
message DeleteOp {
string key = 1;
bytes key = 1;
}
// 4. The Sync Handshake (Vector Clocks)
@@ -63,7 +90,7 @@ message SyncState {
message Frontier {
bytes author_id = 1; // Ed25519 public key (32 bytes)
uint64 max_seq = 2; // Highest sequence number seen from this author
bytes last_hash = 3; // Hash of the last entry (for chain verification)
repeated bytes head_hashes = 3; // All head hashes for this author
}
// 5. Log File Record (wrapper for storage)
@@ -71,3 +98,46 @@ message LogRecord {
bytes hash = 1; // BLAKE3 hash of entry_bytes (32 bytes)
bytes entry_bytes = 2; // Serialized SignedEntry
}
// 6. Join Protocol Messages (new node joining existing mesh)
message JoinRequest {
bytes node_pubkey = 1; // Joining node's public key (32 bytes)
}
message JoinResponse {
bytes store_uuid = 1; // Root store UUID (16 bytes) for new node to create
bytes inviter_pubkey = 2; // Inviter's public key for verification
}
// 7. Sync Protocol Messages (bidirectional sync after join)
message SyncRequest {
bytes store_id = 1; // Store UUID to sync (16 bytes)
SyncState state = 2; // Sender's sync state (for incremental sync)
bool full_sync = 3; // If true, request all entries (for join)
}
message SyncResponse {
bytes store_id = 1; // Store UUID being synced (16 bytes)
SyncState state = 2; // Responder's sync state
}
message SyncEntry {
bytes signed_entry = 1; // Serialized SignedEntry
bytes hash = 2; // Hash for verification
}
message SyncDone {
uint64 entries_sent = 1;
}
// 8. Peer Message Wrapper (for proper message type discrimination)
message PeerMessage {
oneof message {
JoinRequest join_request = 1;
JoinResponse join_response = 2;
SyncRequest sync_request = 3;
SyncResponse sync_response = 4;
SyncEntry sync_entry = 5;
SyncDone sync_done = 6;
}
}