Compare commits
9
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1943e06509 | ||
|
|
4761501ec9 | ||
|
|
c2d4219320 | ||
|
|
a1f134eb02 | ||
|
|
ce852da25e | ||
|
|
57c2906b10 | ||
|
|
346ebccee7 | ||
|
|
f45c6ccfcf | ||
|
|
15dd1b337f |
@@ -4,6 +4,7 @@ members = [
|
|||||||
"lattice-core",
|
"lattice-core",
|
||||||
"lattice-net",
|
"lattice-net",
|
||||||
"lattice-store",
|
"lattice-store",
|
||||||
|
"lattice-cli",
|
||||||
]
|
]
|
||||||
|
|
||||||
[workspace.package]
|
[workspace.package]
|
||||||
@@ -16,6 +17,10 @@ license = "MIT"
|
|||||||
lattice-core = { path = "lattice-core" }
|
lattice-core = { path = "lattice-core" }
|
||||||
lattice-net = { path = "lattice-net" }
|
lattice-net = { path = "lattice-net" }
|
||||||
lattice-store = { path = "lattice-store" }
|
lattice-store = { path = "lattice-store" }
|
||||||
|
lattice-cli = { path = "lattice-cli" }
|
||||||
|
|
||||||
|
# CLI
|
||||||
|
rustyline = "17"
|
||||||
|
|
||||||
# Networking (Iroh)
|
# Networking (Iroh)
|
||||||
iroh = "0.95"
|
iroh = "0.95"
|
||||||
@@ -40,6 +45,8 @@ bytes = "1"
|
|||||||
dirs = "5"
|
dirs = "5"
|
||||||
blake3 = "1"
|
blake3 = "1"
|
||||||
hex = "0.4"
|
hex = "0.4"
|
||||||
|
redb = "2"
|
||||||
|
uuid = { version = "1", features = ["v4"] }
|
||||||
|
|
||||||
# Testing
|
# Testing
|
||||||
tokio-test = "0.4"
|
tokio-test = "0.4"
|
||||||
|
|||||||
+171
-21
@@ -2,20 +2,38 @@
|
|||||||
|
|
||||||
## Ideas
|
## Ideas
|
||||||
|
|
||||||
- SigChains Ed25519-signed, hash-chained append-only logs per node. Trust via local signature verification.
|
**Core:**
|
||||||
- Log-Based State: KV store derived by replaying entries. Watermarks enable safe log pruning + snapshots.
|
- SigChains: Ed25519-signed, hash-chained append-only logs per node.
|
||||||
- Offline-First: Iroh for networking. Vector clocks identify missing entries on reconnect—converges mathematically.
|
- Offline-First: Iroh for networking. Vector clocks identify missing entries on reconnect.
|
||||||
- Full Replication: All nodes keep all logs until watermark consensus, then prune and snapshot.
|
- 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
|
## 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
|
## Stack
|
||||||
|
|
||||||
- rust
|
- rust
|
||||||
- iroh
|
- iroh
|
||||||
- prost protocol buffers
|
- prost protocol buffers
|
||||||
|
- redb (embedded KV store)
|
||||||
|
- rustyline (interactive CLI)
|
||||||
|
|
||||||
### Bootstrap
|
### Bootstrap
|
||||||
|
|
||||||
@@ -58,11 +76,62 @@ Future:
|
|||||||
|
|
||||||
### Data Model
|
### Data Model
|
||||||
|
|
||||||
- Keys are flat strings using path conventions (e.g., `/nodes/{pubkey}`, `/config/sync/interval`).
|
- Multiple KV stores supported, identified by `store_id` (UUID).
|
||||||
- Prefix queries via string matching (sorted map enables efficient range scans).
|
- Keys: Arbitrary byte arrays (`Vec<u8>`), sorted lexicographically.
|
||||||
- State is computed by replaying `Put`/`Delete` operations from all authors.
|
- 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.
|
- 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)
|
### Timestamps (Hybrid Logical Clocks)
|
||||||
|
|
||||||
@@ -86,27 +155,80 @@ Authors apply their own entries through the standard receive path to ensure cons
|
|||||||
Each node stores logs as one file per author:
|
Each node stores logs as one file per author:
|
||||||
|
|
||||||
```
|
```
|
||||||
data/
|
~/.local/share/lattice/
|
||||||
├── identity.key # Local node's Ed25519 private key
|
├── identity.key # Ed25519 private key
|
||||||
├── logs/
|
├── stores/
|
||||||
│ └── {author_id_hex}.log # Append-only SignedEntry stream per author
|
│ └── {store_uuid}/
|
||||||
└── state.db # redb: KV snapshot + vector clocks + indexes
|
│ ├── 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.
|
- Logs: Append-only binary files per `(store, author)`, containing serialized `SignedEntry` messages.
|
||||||
- State DB (redb): Combined KV state, vector clocks, and indexes. Updated as entries are applied.
|
- 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
|
Table Key Value Purpose
|
||||||
─────────────────────────────────────────────────────────────────────────────
|
─────────────────────────────────────────────────────────────────────────────
|
||||||
kv String (path) Vec<u8> Replicated key-value data
|
kv Vec<u8> (key) Vec<HeadInfo> Current tips for each key
|
||||||
vector_clocks [u8; 32] (author_id) (u64 seq, [u8; 32] hash) Track sync state + chain verification
|
applied_frontiers [u8; 32] (author_id) (u64 seq, [u8; 32] hash) What's applied to this store
|
||||||
entry_index (author_id, seq) u64 (offset) Fast entry lookup by position
|
meta Vec<u8> Vec<u8> Store metadata (incl. merkle_root)
|
||||||
meta String Vec<u8> System metadata (own_seq, watermark, etc.)
|
|
||||||
```
|
```
|
||||||
|
|
||||||
|
`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
|
### Watermarks
|
||||||
|
|
||||||
- Nodes gossip their watermarks periodically (throttled).
|
- Nodes gossip their watermarks periodically (throttled).
|
||||||
@@ -114,6 +236,11 @@ meta String Vec<u8> System metada
|
|||||||
- All nodes keep all logs (own + others) for redundancy until watermark consensus.
|
- 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.
|
- 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.
|
- 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)
|
### Rich CRDTs (Future)
|
||||||
|
|
||||||
@@ -157,3 +284,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.
|
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.
|
||||||
+71
-13
@@ -12,7 +12,7 @@
|
|||||||
- [x] Log file I/O (append, read, hash verification)
|
- [x] Log file I/O (append, read, hash verification)
|
||||||
- [x] SigChain (validate entries before appending)
|
- [x] SigChain (validate entries before appending)
|
||||||
- [x] Store (redb) — `kv` + `meta` tables, log replay
|
- [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
|
### Success Criteria
|
||||||
|
|
||||||
@@ -21,15 +21,63 @@
|
|||||||
- Can replay log to reconstruct KV state
|
- Can replay log to reconstruct KV state
|
||||||
- All operations survive restart
|
- All operations survive restart
|
||||||
|
|
||||||
### Multi-KV Refactoring (before M2)
|
### Multi-KV Refactoring (before M2) ✓
|
||||||
|
|
||||||
Current code assumes single store. Changes needed:
|
- [x] DataDir → `stores/{uuid}/` subdirectories
|
||||||
- [ ] DataDir → support `stores/{uuid}/` subdirectories
|
- [x] Store → per-store state.db
|
||||||
- [ ] SigChain → scoped to (store_id, author_id)
|
- [x] Log paths → `stores/{uuid}/logs/{author}.log`
|
||||||
- [ ] Store → per-store state.db, not global
|
- [x] Proto: Entry has store_id (UUID)
|
||||||
- [ ] Log paths → `stores/{uuid}/logs/{author}.log`
|
- [x] CLI → `init`, `create-store`, `list-stores`, `use`
|
||||||
- [ ] Add global meta.db for stores table
|
- [x] meta.db stores table (MetaStore)
|
||||||
- [ ] CLI → `create-store`, `list-stores`, `use <store>`
|
- [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,12 +87,17 @@ Current code assumes single store. Changes needed:
|
|||||||
|
|
||||||
### Deliverables
|
### Deliverables
|
||||||
|
|
||||||
- [ ] Store: add `applied_frontiers` table (sync state per author)
|
**Phase 1: Sync Logic (no network)** ✓
|
||||||
- [ ] VectorClock module (diff, merge, missing entries)
|
- [x] SyncState with AuthorInfo (seq + hash) for hash-based log resumption
|
||||||
- [ ] Sync protocol (push missing entries)
|
- [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**
|
||||||
- [ ] Iroh integration (peer discovery, connection)
|
- [ ] Iroh integration (peer discovery, connection)
|
||||||
- [ ] Multi-author log merging
|
- [ ] Sync protocol (push missing entries over network)
|
||||||
- [ ] CLI: `peers`, `connect`/`join` commands
|
- [ ] CLI: `peers`, `connect`/`join` commands
|
||||||
|
- [ ] Background sync task (tokio::spawn)
|
||||||
|
|
||||||
### Success Criteria
|
### Success Criteria
|
||||||
|
|
||||||
@@ -74,3 +127,8 @@ Current code assumes single store. Changes needed:
|
|||||||
- Snapshots for fast bootstrap
|
- Snapshots for fast bootstrap
|
||||||
- FUSE filesystem mount
|
- FUSE filesystem mount
|
||||||
- Note: FUSE requires u64 inode numbers → maintain `BiMap<u64, Hash>` in redb
|
- 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)
|
||||||
|
|||||||
@@ -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`.
|
||||||
@@ -0,0 +1,18 @@
|
|||||||
|
[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 }
|
||||||
|
rustyline = { workspace = true }
|
||||||
|
hex = { workspace = true }
|
||||||
|
thiserror = { workspace = true }
|
||||||
|
tokio = { workspace = true }
|
||||||
|
shlex = "1"
|
||||||
@@ -0,0 +1,414 @@
|
|||||||
|
//! CLI command handlers
|
||||||
|
|
||||||
|
use crate::node::{LatticeNode, StoreHandle};
|
||||||
|
use lattice_core::Uuid;
|
||||||
|
use std::time::Instant;
|
||||||
|
|
||||||
|
/// Result of a command that may switch stores
|
||||||
|
pub enum CommandResult {
|
||||||
|
/// No store change
|
||||||
|
Ok,
|
||||||
|
/// Switch to this store
|
||||||
|
SwitchTo(StoreHandle),
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Helper to call async code from sync command handlers
|
||||||
|
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(&LatticeNode, Option<&StoreHandle>, &[String]) -> CommandResult;
|
||||||
|
|
||||||
|
pub struct Command {
|
||||||
|
pub name: &'static str,
|
||||||
|
pub args: &'static str,
|
||||||
|
pub description: &'static str,
|
||||||
|
pub min_args: usize,
|
||||||
|
pub max_args: usize,
|
||||||
|
pub handler: Handler,
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn commands() -> Vec<Command> {
|
||||||
|
vec![
|
||||||
|
Command {
|
||||||
|
name: "init",
|
||||||
|
args: "",
|
||||||
|
description: "Initialize node with root store",
|
||||||
|
min_args: 0,
|
||||||
|
max_args: 0,
|
||||||
|
handler: cmd_init,
|
||||||
|
},
|
||||||
|
Command {
|
||||||
|
name: "create-store",
|
||||||
|
args: "",
|
||||||
|
description: "Create a new store",
|
||||||
|
min_args: 0,
|
||||||
|
max_args: 0,
|
||||||
|
handler: cmd_create_store,
|
||||||
|
},
|
||||||
|
Command {
|
||||||
|
name: "use",
|
||||||
|
args: "<uuid>",
|
||||||
|
description: "Switch to a store",
|
||||||
|
min_args: 1,
|
||||||
|
max_args: 1,
|
||||||
|
handler: cmd_use_store,
|
||||||
|
},
|
||||||
|
Command {
|
||||||
|
name: "list-stores",
|
||||||
|
args: "",
|
||||||
|
description: "List all stores",
|
||||||
|
min_args: 0,
|
||||||
|
max_args: 0,
|
||||||
|
handler: cmd_list_stores,
|
||||||
|
},
|
||||||
|
Command {
|
||||||
|
name: "put",
|
||||||
|
args: "<key> <value>",
|
||||||
|
description: "Store a key-value pair",
|
||||||
|
min_args: 2,
|
||||||
|
max_args: 2,
|
||||||
|
handler: cmd_put,
|
||||||
|
},
|
||||||
|
Command {
|
||||||
|
name: "get",
|
||||||
|
args: "<key> [-v]",
|
||||||
|
description: "Retrieve a value by key",
|
||||||
|
min_args: 1,
|
||||||
|
max_args: 2,
|
||||||
|
handler: cmd_get,
|
||||||
|
},
|
||||||
|
Command {
|
||||||
|
name: "delete",
|
||||||
|
args: "<key>",
|
||||||
|
description: "Delete a key",
|
||||||
|
min_args: 1,
|
||||||
|
max_args: 1,
|
||||||
|
handler: cmd_delete,
|
||||||
|
},
|
||||||
|
Command {
|
||||||
|
name: "list",
|
||||||
|
args: "[-v]",
|
||||||
|
description: "List all key-value pairs (-v for verbose)",
|
||||||
|
min_args: 0,
|
||||||
|
max_args: 1,
|
||||||
|
handler: cmd_list,
|
||||||
|
},
|
||||||
|
Command {
|
||||||
|
name: "status",
|
||||||
|
args: "",
|
||||||
|
description: "Show node/store info",
|
||||||
|
min_args: 0,
|
||||||
|
max_args: 0,
|
||||||
|
handler: cmd_status,
|
||||||
|
},
|
||||||
|
Command {
|
||||||
|
name: "author-state",
|
||||||
|
args: "[author-hex]",
|
||||||
|
description: "Show author state (default: self)",
|
||||||
|
min_args: 0,
|
||||||
|
max_args: 1,
|
||||||
|
handler: cmd_author_state,
|
||||||
|
},
|
||||||
|
Command {
|
||||||
|
name: "help",
|
||||||
|
args: "",
|
||||||
|
description: "Show this help message",
|
||||||
|
min_args: 0,
|
||||||
|
max_args: 0,
|
||||||
|
handler: cmd_help,
|
||||||
|
},
|
||||||
|
]
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- Store management ---
|
||||||
|
|
||||||
|
fn cmd_init(node: &LatticeNode, _store: Option<&StoreHandle>, _args: &[String]) -> CommandResult {
|
||||||
|
match node.init() {
|
||||||
|
Ok(store_id) => {
|
||||||
|
println!("Initialized with root store: {}", store_id);
|
||||||
|
match node.open_store(store_id) {
|
||||||
|
Ok((handle, _)) => CommandResult::SwitchTo(handle),
|
||||||
|
Err(e) => {
|
||||||
|
eprintln!("Warning: {}", e);
|
||||||
|
CommandResult::Ok
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
eprintln!("Error: {}", e);
|
||||||
|
CommandResult::Ok
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn cmd_create_store(node: &LatticeNode, _store: Option<&StoreHandle>, _args: &[String]) -> CommandResult {
|
||||||
|
match node.create_store() {
|
||||||
|
Ok(store_id) => {
|
||||||
|
println!("Created store: {}", store_id);
|
||||||
|
match 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: &LatticeNode, _store: Option<&StoreHandle>, 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 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: &LatticeNode, store: Option<&StoreHandle>, _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_help(_node: &LatticeNode, _store: Option<&StoreHandle>, _args: &[String]) -> CommandResult {
|
||||||
|
println!("\nCommands:");
|
||||||
|
for cmd in commands() {
|
||||||
|
if cmd.args.is_empty() {
|
||||||
|
println!(" {:<16} {}", cmd.name, cmd.description);
|
||||||
|
} else {
|
||||||
|
println!(" {} {:<8} {}", cmd.name, cmd.args, cmd.description);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
println!(" quit Exit");
|
||||||
|
println!();
|
||||||
|
CommandResult::Ok
|
||||||
|
}
|
||||||
|
|
||||||
|
fn cmd_status(node: &LatticeNode, store: Option<&StoreHandle>, _args: &[String]) -> CommandResult {
|
||||||
|
println!("Node ID: {}", hex::encode(node.node_id()));
|
||||||
|
println!("Data: {}", node.data_path().display());
|
||||||
|
match node.root_store() {
|
||||||
|
Ok(Some(id)) => println!("Root: {}", id),
|
||||||
|
Ok(None) => println!("Root: (not set)"),
|
||||||
|
Err(_) => println!("Root: (error)"),
|
||||||
|
}
|
||||||
|
if let Some(h) = store {
|
||||||
|
println!("Store: {}", h.id());
|
||||||
|
println!("Log Seq: {}", block_async(h.log_seq()));
|
||||||
|
println!("Applied: {}", block_async(h.applied_seq()).unwrap_or(0));
|
||||||
|
} else {
|
||||||
|
println!("Store: (none)");
|
||||||
|
}
|
||||||
|
CommandResult::Ok
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- KV ---
|
||||||
|
|
||||||
|
fn cmd_put(_node: &LatticeNode, store: Option<&StoreHandle>, 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: &LatticeNode, store: Option<&StoreHandle>, 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: &LatticeNode, store: Option<&StoreHandle>, 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: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) -> CommandResult {
|
||||||
|
let Some(h) = store else {
|
||||||
|
println!("No store selected. Use 'init' or 'use <uuid>'");
|
||||||
|
return CommandResult::Ok;
|
||||||
|
};
|
||||||
|
let verbose = args.first().map(|a| a == "-v").unwrap_or(false);
|
||||||
|
let start = Instant::now();
|
||||||
|
match block_async(h.list()) {
|
||||||
|
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 {
|
||||||
|
println!("{} = {}", key_str, format_value(v));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
println!("({} keys, {:.2?})", entries.len(), start.elapsed());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
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)))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn cmd_author_state(node: &LatticeNode, store: Option<&StoreHandle>, 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
|
||||||
|
}
|
||||||
@@ -0,0 +1,108 @@
|
|||||||
|
//! Lattice Interactive CLI
|
||||||
|
|
||||||
|
mod node;
|
||||||
|
mod commands;
|
||||||
|
mod store_actor;
|
||||||
|
|
||||||
|
use commands::CommandResult;
|
||||||
|
use node::{LatticeNodeBuilder, StoreHandle};
|
||||||
|
use rustyline::error::ReadlineError;
|
||||||
|
use rustyline::DefaultEditor;
|
||||||
|
|
||||||
|
#[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 LatticeNodeBuilder::new().build() {
|
||||||
|
Ok(n) => n,
|
||||||
|
Err(e) => {
|
||||||
|
eprintln!("Failed to initialize: {}", e);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
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() {
|
||||||
|
Ok(Some((h, 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);
|
||||||
|
}
|
||||||
|
Some(h)
|
||||||
|
}
|
||||||
|
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 ¤t_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("");
|
||||||
|
|
||||||
|
if cmd_name == "quit" || cmd_name == "exit" {
|
||||||
|
println!("Goodbye!");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
match cmds.iter().find(|c| c.name == cmd_name) {
|
||||||
|
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(), cmd_args) {
|
||||||
|
CommandResult::Ok => {}
|
||||||
|
CommandResult::SwitchTo(h) => current_store = Some(h),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
None => println!("Unknown: '{}'. Type 'help'.", cmd_name),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(ReadlineError::Interrupted | ReadlineError::Eof) => {
|
||||||
|
println!("Goodbye!");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
eprintln!("Error: {:?}", e);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,433 @@
|
|||||||
|
//! Local Lattice node API with multi-store support
|
||||||
|
|
||||||
|
use lattice_core::{
|
||||||
|
DataDir, MetaStore, Node, SigChain, Store, Uuid,
|
||||||
|
log::LogError,
|
||||||
|
meta_store::MetaStoreError,
|
||||||
|
sigchain::SigChainError,
|
||||||
|
store::StoreError,
|
||||||
|
};
|
||||||
|
use std::path::Path;
|
||||||
|
use std::rc::Rc;
|
||||||
|
use thiserror::Error;
|
||||||
|
|
||||||
|
#[derive(Error, Debug)]
|
||||||
|
pub enum NodeError {
|
||||||
|
#[error("IO error: {0}")]
|
||||||
|
Io(#[from] std::io::Error),
|
||||||
|
|
||||||
|
#[error("Store error: {0}")]
|
||||||
|
Store(#[from] StoreError),
|
||||||
|
|
||||||
|
#[error("MetaStore error: {0}")]
|
||||||
|
MetaStore(#[from] MetaStoreError),
|
||||||
|
|
||||||
|
#[error("SigChain error: {0}")]
|
||||||
|
SigChain(#[from] SigChainError),
|
||||||
|
|
||||||
|
#[error("Log error: {0}")]
|
||||||
|
Log(#[from] LogError),
|
||||||
|
|
||||||
|
#[error("Node error: {0}")]
|
||||||
|
Node(#[from] lattice_core::node::NodeError),
|
||||||
|
|
||||||
|
#[error("Already initialized")]
|
||||||
|
AlreadyInitialized,
|
||||||
|
|
||||||
|
#[error("Channel closed")]
|
||||||
|
ChannelClosed,
|
||||||
|
|
||||||
|
#[error("Actor error: {0}")]
|
||||||
|
Actor(String),
|
||||||
|
}
|
||||||
|
|
||||||
|
pub struct NodeInfo {
|
||||||
|
pub node_id: String,
|
||||||
|
pub data_path: String,
|
||||||
|
pub stores: Vec<Uuid>,
|
||||||
|
}
|
||||||
|
|
||||||
|
pub struct StoreInfo {
|
||||||
|
pub store_id: Uuid,
|
||||||
|
pub entries_replayed: u64,
|
||||||
|
}
|
||||||
|
|
||||||
|
pub struct LatticeNodeBuilder {
|
||||||
|
pub data_dir: DataDir,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl LatticeNodeBuilder {
|
||||||
|
pub fn new() -> Self {
|
||||||
|
Self { data_dir: DataDir::default() }
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn build(self) -> Result<LatticeNode, NodeError> {
|
||||||
|
self.data_dir.ensure_dirs()?;
|
||||||
|
|
||||||
|
let key_path = self.data_dir.identity_key();
|
||||||
|
let node = if key_path.exists() {
|
||||||
|
Node::load(&key_path)?
|
||||||
|
} else {
|
||||||
|
let node = Node::generate();
|
||||||
|
node.save(&key_path)?;
|
||||||
|
node
|
||||||
|
};
|
||||||
|
|
||||||
|
let meta = MetaStore::open(self.data_dir.meta_db())?;
|
||||||
|
|
||||||
|
Ok(LatticeNode {
|
||||||
|
data_dir: self.data_dir,
|
||||||
|
node: Rc::new(node),
|
||||||
|
meta,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Default for LatticeNodeBuilder {
|
||||||
|
fn default() -> Self { Self::new() }
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A local Lattice node (manages identity and store registry)
|
||||||
|
pub struct LatticeNode {
|
||||||
|
data_dir: DataDir,
|
||||||
|
node: Rc<Node>,
|
||||||
|
meta: MetaStore,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl LatticeNode {
|
||||||
|
pub fn info(&self) -> NodeInfo {
|
||||||
|
NodeInfo {
|
||||||
|
node_id: hex::encode(self.node.public_key_bytes()),
|
||||||
|
data_path: self.data_dir.base().display().to_string(),
|
||||||
|
stores: self.meta.list_stores().unwrap_or_default(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn node_id(&self) -> [u8; 32] {
|
||||||
|
self.node.public_key_bytes()
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn data_path(&self) -> &Path {
|
||||||
|
self.data_dir.base()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Get the root store ID
|
||||||
|
pub fn root_store(&self) -> Result<Option<Uuid>, NodeError> {
|
||||||
|
Ok(self.meta.root_store()?)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Open the root store if set
|
||||||
|
pub fn open_root_store(&self) -> Result<Option<(StoreHandle, StoreInfo)>, NodeError> {
|
||||||
|
match self.meta.root_store()? {
|
||||||
|
Some(id) => Ok(Some(self.open_store(id)?)),
|
||||||
|
None => Ok(None),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Initialize the node with a root store (fails if already initialized)
|
||||||
|
pub fn init(&self) -> Result<Uuid, NodeError> {
|
||||||
|
if self.meta.root_store()?.is_some() {
|
||||||
|
return Err(NodeError::AlreadyInitialized);
|
||||||
|
}
|
||||||
|
let store_id = self.create_store()?;
|
||||||
|
self.meta.set_root_store(store_id)?;
|
||||||
|
Ok(store_id)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn list_stores(&self) -> Result<Vec<Uuid>, NodeError> {
|
||||||
|
Ok(self.meta.list_stores()?)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn create_store(&self) -> Result<Uuid, NodeError> {
|
||||||
|
let store_id = Uuid::new_v4();
|
||||||
|
self.data_dir.ensure_store_dirs(store_id)?;
|
||||||
|
let _ = Store::open(self.data_dir.store_state_db(store_id))?;
|
||||||
|
self.meta.add_store(store_id)?;
|
||||||
|
Ok(store_id)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn open_store(&self, store_id: Uuid) -> Result<(StoreHandle, StoreInfo), NodeError> {
|
||||||
|
self.data_dir.ensure_store_dirs(store_id)?;
|
||||||
|
|
||||||
|
let author_id_hex = hex::encode(self.node.public_key_bytes());
|
||||||
|
let log_path = self.data_dir.store_log_file(store_id, &author_id_hex);
|
||||||
|
|
||||||
|
let sigchain = if log_path.exists() {
|
||||||
|
SigChain::from_log(&log_path, *store_id.as_bytes(), self.node.public_key_bytes())?
|
||||||
|
} else {
|
||||||
|
SigChain::new(&log_path, *store_id.as_bytes(), self.node.public_key_bytes())
|
||||||
|
};
|
||||||
|
|
||||||
|
let store = Store::open(self.data_dir.store_state_db(store_id))?;
|
||||||
|
let entries_replayed = if log_path.exists() {
|
||||||
|
store.replay_log(&log_path)?
|
||||||
|
} else {
|
||||||
|
0
|
||||||
|
};
|
||||||
|
|
||||||
|
let info = StoreInfo { store_id, entries_replayed };
|
||||||
|
|
||||||
|
// Spawn actor thread - actor owns store, sigchain, and node copy
|
||||||
|
let (tx, actor_handle) = crate::store_actor::spawn_store_actor(
|
||||||
|
store_id,
|
||||||
|
store,
|
||||||
|
sigchain,
|
||||||
|
(*self.node).clone(),
|
||||||
|
);
|
||||||
|
|
||||||
|
let handle = StoreHandle {
|
||||||
|
store_id,
|
||||||
|
tx,
|
||||||
|
actor_handle: Some(actor_handle),
|
||||||
|
};
|
||||||
|
|
||||||
|
Ok((handle, info))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A handle to a specific store - wraps channel to actor thread
|
||||||
|
pub struct StoreHandle {
|
||||||
|
store_id: Uuid,
|
||||||
|
tx: tokio::sync::mpsc::Sender<crate::store_actor::StoreCmd>,
|
||||||
|
actor_handle: Option<std::thread::JoinHandle<()>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl StoreHandle {
|
||||||
|
pub fn id(&self) -> Uuid { self.store_id }
|
||||||
|
|
||||||
|
pub async fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>, NodeError> {
|
||||||
|
use crate::store_actor::StoreCmd;
|
||||||
|
let (resp_tx, resp_rx) = tokio::sync::oneshot::channel();
|
||||||
|
self.tx.send(StoreCmd::Get { key: key.to_vec(), resp: resp_tx }).await
|
||||||
|
.map_err(|_| NodeError::ChannelClosed)?;
|
||||||
|
resp_rx.await
|
||||||
|
.map_err(|_| NodeError::ChannelClosed)?
|
||||||
|
.map_err(NodeError::Store)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn get_heads(&self, key: &[u8]) -> Result<Vec<lattice_core::HeadInfo>, NodeError> {
|
||||||
|
use crate::store_actor::StoreCmd;
|
||||||
|
let (resp_tx, resp_rx) = tokio::sync::oneshot::channel();
|
||||||
|
self.tx.send(StoreCmd::GetHeads { key: key.to_vec(), resp: resp_tx }).await
|
||||||
|
.map_err(|_| NodeError::ChannelClosed)?;
|
||||||
|
resp_rx.await
|
||||||
|
.map_err(|_| NodeError::ChannelClosed)?
|
||||||
|
.map_err(NodeError::Store)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn list(&self) -> Result<Vec<(Vec<u8>, Vec<u8>)>, NodeError> {
|
||||||
|
use crate::store_actor::StoreCmd;
|
||||||
|
let (resp_tx, resp_rx) = tokio::sync::oneshot::channel();
|
||||||
|
self.tx.send(StoreCmd::List { resp: resp_tx }).await
|
||||||
|
.map_err(|_| NodeError::ChannelClosed)?;
|
||||||
|
resp_rx.await
|
||||||
|
.map_err(|_| NodeError::ChannelClosed)?
|
||||||
|
.map_err(NodeError::Store)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn log_seq(&self) -> u64 {
|
||||||
|
use crate::store_actor::StoreCmd;
|
||||||
|
let (resp_tx, resp_rx) = tokio::sync::oneshot::channel();
|
||||||
|
let _ = self.tx.send(StoreCmd::LogSeq { resp: resp_tx }).await;
|
||||||
|
resp_rx.await.unwrap_or(0)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn applied_seq(&self) -> Result<u64, NodeError> {
|
||||||
|
use crate::store_actor::StoreCmd;
|
||||||
|
let (resp_tx, resp_rx) = tokio::sync::oneshot::channel();
|
||||||
|
self.tx.send(StoreCmd::AppliedSeq { resp: resp_tx }).await
|
||||||
|
.map_err(|_| NodeError::ChannelClosed)?;
|
||||||
|
resp_rx.await
|
||||||
|
.map_err(|_| NodeError::ChannelClosed)?
|
||||||
|
.map_err(NodeError::Store)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn author_state(&self, author: &[u8; 32]) -> Result<Option<lattice_core::proto::AuthorState>, NodeError> {
|
||||||
|
use crate::store_actor::StoreCmd;
|
||||||
|
let (resp_tx, resp_rx) = tokio::sync::oneshot::channel();
|
||||||
|
self.tx.send(StoreCmd::AuthorState { author: *author, resp: resp_tx }).await
|
||||||
|
.map_err(|_| NodeError::ChannelClosed)?;
|
||||||
|
resp_rx.await
|
||||||
|
.map_err(|_| NodeError::ChannelClosed)?
|
||||||
|
.map_err(NodeError::Store)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn put(&self, key: &[u8], value: &[u8]) -> Result<u64, NodeError> {
|
||||||
|
use crate::store_actor::StoreCmd;
|
||||||
|
let (resp_tx, resp_rx) = tokio::sync::oneshot::channel();
|
||||||
|
self.tx.send(StoreCmd::Put { key: key.to_vec(), value: value.to_vec(), resp: resp_tx }).await
|
||||||
|
.map_err(|_| NodeError::ChannelClosed)?;
|
||||||
|
resp_rx.await
|
||||||
|
.map_err(|_| NodeError::ChannelClosed)?
|
||||||
|
.map_err(|e| NodeError::Actor(e.to_string()))
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn delete(&self, key: &[u8]) -> Result<u64, NodeError> {
|
||||||
|
use crate::store_actor::StoreCmd;
|
||||||
|
let (resp_tx, resp_rx) = tokio::sync::oneshot::channel();
|
||||||
|
self.tx.send(StoreCmd::Delete { key: key.to_vec(), resp: resp_tx }).await
|
||||||
|
.map_err(|_| NodeError::ChannelClosed)?;
|
||||||
|
resp_rx.await
|
||||||
|
.map_err(|_| NodeError::ChannelClosed)?
|
||||||
|
.map_err(|e| NodeError::Actor(e.to_string()))
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Drop for StoreHandle {
|
||||||
|
fn drop(&mut self) {
|
||||||
|
// Send shutdown command (non-blocking) and wait for actor to finish
|
||||||
|
// Use try_send to avoid panic in async context
|
||||||
|
let _ = self.tx.try_send(crate::store_actor::StoreCmd::Shutdown);
|
||||||
|
if let Some(handle) = self.actor_handle.take() {
|
||||||
|
let _ = handle.join();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use std::env::temp_dir;
|
||||||
|
|
||||||
|
fn temp_data_dir(name: &str) -> DataDir {
|
||||||
|
let path = temp_dir().join(format!("lattice_node_test_{}", name));
|
||||||
|
let _ = std::fs::remove_dir_all(&path);
|
||||||
|
DataDir::new(path)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_create_and_open_store() {
|
||||||
|
let data_dir = temp_data_dir("meta_store");
|
||||||
|
|
||||||
|
let node = LatticeNodeBuilder { data_dir: data_dir.clone() }
|
||||||
|
.build()
|
||||||
|
.expect("Failed to create node");
|
||||||
|
|
||||||
|
assert!(node.info().stores.is_empty());
|
||||||
|
|
||||||
|
let store_id = node.create_store().expect("Failed to create store");
|
||||||
|
|
||||||
|
// Verify it's in the list
|
||||||
|
let stores = node.list_stores().expect("list failed");
|
||||||
|
assert!(stores.contains(&store_id));
|
||||||
|
|
||||||
|
let (handle, _) = node.open_store(store_id).expect("Failed to open store");
|
||||||
|
handle.put(b"/key", b"value").await.expect("put failed");
|
||||||
|
assert_eq!(handle.get(b"/key").await.unwrap(), Some(b"value".to_vec()));
|
||||||
|
|
||||||
|
let _ = std::fs::remove_dir_all(data_dir.base());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_store_isolation() {
|
||||||
|
let data_dir = temp_data_dir("meta_isolation");
|
||||||
|
|
||||||
|
let node = LatticeNodeBuilder { data_dir: data_dir.clone() }
|
||||||
|
.build()
|
||||||
|
.expect("Failed to create node");
|
||||||
|
|
||||||
|
let store_a = node.create_store().expect("create A");
|
||||||
|
let store_b = node.create_store().expect("create B");
|
||||||
|
|
||||||
|
let (handle_a, _) = node.open_store(store_a).expect("open A");
|
||||||
|
handle_a.put(b"/key", b"from A").await.expect("put A");
|
||||||
|
|
||||||
|
let (handle_b, _) = node.open_store(store_b).expect("open B");
|
||||||
|
assert_eq!(handle_b.get(b"/key").await.unwrap(), None);
|
||||||
|
|
||||||
|
assert_eq!(handle_a.get(b"/key").await.unwrap(), Some(b"from A".to_vec()));
|
||||||
|
|
||||||
|
let _ = std::fs::remove_dir_all(data_dir.base());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_init_creates_root_store() {
|
||||||
|
let data_dir = temp_data_dir("init_root");
|
||||||
|
|
||||||
|
let node = LatticeNodeBuilder { data_dir: data_dir.clone() }
|
||||||
|
.build()
|
||||||
|
.expect("create node");
|
||||||
|
|
||||||
|
// Initially no root store
|
||||||
|
assert!(node.root_store().unwrap().is_none());
|
||||||
|
|
||||||
|
// Init creates root store
|
||||||
|
let root_id = node.init().expect("init failed");
|
||||||
|
assert_eq!(node.root_store().unwrap(), Some(root_id));
|
||||||
|
|
||||||
|
let _ = std::fs::remove_dir_all(data_dir.base());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_duplicate_init_fails() {
|
||||||
|
let data_dir = temp_data_dir("init_dup");
|
||||||
|
|
||||||
|
let node = LatticeNodeBuilder { data_dir: data_dir.clone() }
|
||||||
|
.build()
|
||||||
|
.expect("create node");
|
||||||
|
|
||||||
|
node.init().expect("first init");
|
||||||
|
|
||||||
|
// Second init should fail
|
||||||
|
match node.init() {
|
||||||
|
Err(NodeError::AlreadyInitialized) => (),
|
||||||
|
other => panic!("Expected AlreadyInitialized, got {:?}", other),
|
||||||
|
}
|
||||||
|
|
||||||
|
let _ = std::fs::remove_dir_all(data_dir.base());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_root_store_in_info_after_init() {
|
||||||
|
let data_dir = temp_data_dir("init_info");
|
||||||
|
|
||||||
|
// First session: init
|
||||||
|
let root_id = {
|
||||||
|
let node = LatticeNodeBuilder { data_dir: data_dir.clone() }
|
||||||
|
.build()
|
||||||
|
.expect("create node");
|
||||||
|
node.init().expect("init")
|
||||||
|
};
|
||||||
|
|
||||||
|
// Second session: root_store should persist
|
||||||
|
let node = LatticeNodeBuilder { data_dir: data_dir.clone() }
|
||||||
|
.build()
|
||||||
|
.expect("reload node");
|
||||||
|
|
||||||
|
assert_eq!(node.root_store().unwrap(), Some(root_id));
|
||||||
|
|
||||||
|
let _ = std::fs::remove_dir_all(data_dir.base());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_idempotent_put_and_delete() {
|
||||||
|
let data_dir = temp_data_dir("idempotent");
|
||||||
|
|
||||||
|
let node = LatticeNodeBuilder { data_dir: data_dir.clone() }
|
||||||
|
.build()
|
||||||
|
.expect("create node");
|
||||||
|
let store_id = node.init().expect("init");
|
||||||
|
let (store, _) = node.open_store(store_id).expect("open store");
|
||||||
|
|
||||||
|
// Put twice with same value - second should be idempotent
|
||||||
|
let seq1 = store.put(b"/key", b"value").await.expect("put 1");
|
||||||
|
assert_eq!(seq1, 1);
|
||||||
|
|
||||||
|
let seq2 = store.put(b"/key", b"value").await.expect("put 2");
|
||||||
|
assert_eq!(seq2, 1, "Second put with same value should be idempotent (no new entry)");
|
||||||
|
|
||||||
|
assert_eq!(store.log_seq().await, 1, "Log should have 1 entry, not 2");
|
||||||
|
|
||||||
|
// Delete twice - second should be idempotent
|
||||||
|
let seq3 = store.delete(b"/key").await.expect("delete 1");
|
||||||
|
assert_eq!(seq3, 2);
|
||||||
|
|
||||||
|
let seq4 = store.delete(b"/key").await.expect("delete 2");
|
||||||
|
assert_eq!(seq4, 2, "Second delete should be idempotent (no new entry)");
|
||||||
|
|
||||||
|
assert_eq!(store.log_seq().await, 2, "Log should have 2 entries, not 3");
|
||||||
|
|
||||||
|
let _ = std::fs::remove_dir_all(data_dir.base());
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,201 @@
|
|||||||
|
//! Store Actor - dedicated thread that owns Store and processes commands via channel
|
||||||
|
|
||||||
|
use lattice_core::{
|
||||||
|
EntryBuilder, HeadInfo, Node, SigChain, Store, Uuid,
|
||||||
|
hlc::HLC,
|
||||||
|
proto::AuthorState,
|
||||||
|
sigchain::SigChainError,
|
||||||
|
store::StoreError,
|
||||||
|
};
|
||||||
|
use tokio::sync::{mpsc, oneshot};
|
||||||
|
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 {
|
||||||
|
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>>,
|
||||||
|
},
|
||||||
|
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 SigChain
|
||||||
|
pub struct StoreActor {
|
||||||
|
store_id: Uuid,
|
||||||
|
store: Store,
|
||||||
|
sigchain: SigChain,
|
||||||
|
node: Node,
|
||||||
|
rx: mpsc::Receiver<StoreCmd>,
|
||||||
|
}
|
||||||
|
|
||||||
|
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: Node,
|
||||||
|
rx: mpsc::Receiver<StoreCmd>,
|
||||||
|
) -> Self {
|
||||||
|
Self {
|
||||||
|
store_id,
|
||||||
|
store,
|
||||||
|
sigchain,
|
||||||
|
node,
|
||||||
|
rx,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// 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 { resp } => {
|
||||||
|
let _ = resp.send(self.store.list_all());
|
||||||
|
}
|
||||||
|
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 _ = resp.send(self.sigchain.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::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) {
|
||||||
|
return Ok(self.sigchain.len()); // Idempotent, no new entry
|
||||||
|
}
|
||||||
|
|
||||||
|
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) {
|
||||||
|
return Ok(self.sigchain.len()); // Idempotent, no new entry
|
||||||
|
}
|
||||||
|
|
||||||
|
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 seq = self.sigchain.len() + 1;
|
||||||
|
let prev_hash = self.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);
|
||||||
|
|
||||||
|
self.sigchain.append(&entry)?;
|
||||||
|
self.store.apply_entry(&entry)?;
|
||||||
|
|
||||||
|
Ok(seq)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Spawn a store actor in a new thread, returns (sender, join_handle)
|
||||||
|
/// Uses std::thread since redb is blocking
|
||||||
|
pub fn spawn_store_actor(
|
||||||
|
store_id: Uuid,
|
||||||
|
store: Store,
|
||||||
|
sigchain: SigChain,
|
||||||
|
node: Node,
|
||||||
|
) -> (mpsc::Sender<StoreCmd>, JoinHandle<()>) {
|
||||||
|
let (tx, rx) = mpsc::channel(32);
|
||||||
|
let actor = StoreActor::new(store_id, store, sigchain, node, rx);
|
||||||
|
let handle = thread::spawn(move || actor.run());
|
||||||
|
(tx, handle)
|
||||||
|
}
|
||||||
@@ -14,6 +14,8 @@ bytes = { workspace = true }
|
|||||||
dirs = { workspace = true }
|
dirs = { workspace = true }
|
||||||
blake3 = { workspace = true }
|
blake3 = { workspace = true }
|
||||||
hex = { workspace = true }
|
hex = { workspace = true }
|
||||||
|
redb = { workspace = true }
|
||||||
|
uuid = { workspace = true }
|
||||||
|
|
||||||
[build-dependencies]
|
[build-dependencies]
|
||||||
prost-build = { workspace = true }
|
prost-build = { workspace = true }
|
||||||
|
|||||||
@@ -2,19 +2,26 @@
|
|||||||
//!
|
//!
|
||||||
//! Provides platform-specific paths for Lattice data storage:
|
//! Provides platform-specific paths for Lattice data storage:
|
||||||
//! - `identity.key` — Ed25519 private key
|
//! - `identity.key` — Ed25519 private key
|
||||||
//! - `logs/` — Append-only log files per author
|
//! - `meta.db` — Global metadata (stores table)
|
||||||
//! - `state.db` — KV snapshot and indexes
|
//! - `stores/{uuid}/logs/{author}.log` — Per-store, per-author logs
|
||||||
|
//! - `stores/{uuid}/state.db` — Per-store KV state
|
||||||
|
|
||||||
use std::path::{Path, PathBuf};
|
use std::path::{Path, PathBuf};
|
||||||
|
use uuid::Uuid;
|
||||||
|
|
||||||
const APP_NAME: &str = "lattice";
|
const APP_NAME: &str = "lattice";
|
||||||
|
|
||||||
/// Data directory configuration.
|
/// Data directory configuration.
|
||||||
///
|
///
|
||||||
/// Handles paths for:
|
/// Multi-store layout:
|
||||||
/// - `identity.key` — node's private key
|
/// ```text
|
||||||
/// - `logs/{author_id}.log` — per-author log files
|
/// base/
|
||||||
/// - `state.db` — redb database
|
/// identity.key
|
||||||
|
/// meta.db
|
||||||
|
/// stores/{uuid}/
|
||||||
|
/// logs/{author}.log
|
||||||
|
/// state.db
|
||||||
|
/// ```
|
||||||
#[derive(Debug, Clone)]
|
#[derive(Debug, Clone)]
|
||||||
pub struct DataDir {
|
pub struct DataDir {
|
||||||
base: PathBuf,
|
base: PathBuf,
|
||||||
@@ -27,10 +34,6 @@ impl DataDir {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Create a DataDir using the platform-specific data directory.
|
/// 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> {
|
pub fn default_location() -> Option<Self> {
|
||||||
dirs::data_dir().map(|d| Self::new(d.join(APP_NAME)))
|
dirs::data_dir().map(|d| Self::new(d.join(APP_NAME)))
|
||||||
}
|
}
|
||||||
@@ -45,25 +48,47 @@ impl DataDir {
|
|||||||
self.base.join("identity.key")
|
self.base.join("identity.key")
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Get the path to the logs directory.
|
/// Get the path to the global metadata database.
|
||||||
pub fn logs_dir(&self) -> PathBuf {
|
pub fn meta_db(&self) -> PathBuf {
|
||||||
self.base.join("logs")
|
self.base.join("meta.db")
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Get the path to a specific author's log file.
|
/// Get the path to the stores directory.
|
||||||
pub fn log_file(&self, author_id_hex: &str) -> PathBuf {
|
pub fn stores_dir(&self) -> PathBuf {
|
||||||
self.logs_dir().join(format!("{}.log", author_id_hex))
|
self.base.join("stores")
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Get the path to the state database.
|
/// Get the path to a specific store's directory.
|
||||||
pub fn state_db(&self) -> PathBuf {
|
pub fn store_dir(&self, store_id: Uuid) -> PathBuf {
|
||||||
self.base.join("state.db")
|
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<()> {
|
pub fn ensure_dirs(&self) -> std::io::Result<()> {
|
||||||
std::fs::create_dir_all(&self.base)?;
|
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(())
|
Ok(())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -83,29 +108,31 @@ mod tests {
|
|||||||
let dd = DataDir::new("/custom/path");
|
let dd = DataDir::new("/custom/path");
|
||||||
assert_eq!(dd.base(), Path::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.identity_key(), PathBuf::from("/custom/path/identity.key"));
|
||||||
assert_eq!(dd.logs_dir(), PathBuf::from("/custom/path/logs"));
|
assert_eq!(dd.meta_db(), PathBuf::from("/custom/path/meta.db"));
|
||||||
assert_eq!(dd.state_db(), PathBuf::from("/custom/path/state.db"));
|
assert_eq!(dd.stores_dir(), PathBuf::from("/custom/path/stores"));
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_log_file_path() {
|
fn test_store_paths() {
|
||||||
let dd = DataDir::new("/data");
|
let dd = DataDir::new("/data");
|
||||||
let path = dd.log_file("abc123");
|
let store_id = Uuid::parse_str("a1b2c3d4-e5f6-7890-abcd-ef1234567890").unwrap();
|
||||||
assert_eq!(path, PathBuf::from("/data/logs/abc123.log"));
|
|
||||||
|
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]
|
#[test]
|
||||||
fn test_default_location_exists() {
|
fn test_default_location_exists() {
|
||||||
// On most systems, default_location should return Some
|
// On most systems, default_location should return Some
|
||||||
let location = DataDir::default_location();
|
let location = DataDir::default_location();
|
||||||
// Just verify it doesn't panic - actual path varies by platform
|
|
||||||
assert!(location.is_some() || true);
|
assert!(location.is_some() || true);
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_default_impl() {
|
fn test_default_impl() {
|
||||||
let dd = DataDir::default();
|
let dd = DataDir::default();
|
||||||
// Should either be platform default or ./data fallback
|
|
||||||
assert!(dd.base().to_str().is_some());
|
assert!(dd.base().to_str().is_some());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+11
-4
@@ -4,24 +4,27 @@
|
|||||||
//! - **Node**: Identity with Ed25519 keypair
|
//! - **Node**: Identity with Ed25519 keypair
|
||||||
//! - **SigChain**: Append-only cryptographically signed log
|
//! - **SigChain**: Append-only cryptographically signed log
|
||||||
//! - **Entry**: Atomic operations in the 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
|
//! - **HLC**: Hybrid Logical Clock for ordering
|
||||||
//! - **Clock**: Time abstraction for testability
|
//! - **Clock**: Time abstraction for testability
|
||||||
//! - **Proto**: Generated protobuf types from lattice.proto
|
//! - **Proto**: Generated protobuf types from lattice.proto
|
||||||
//! - **DataDir**: Platform-specific data directory paths
|
//! - **DataDir**: Platform-specific data directory paths
|
||||||
//! - **SignedEntry**: Entry creation, signing, and verification
|
//! - **SignedEntry**: Entry creation, signing, and verification
|
||||||
//! - **Log**: Append-only log file I/O
|
//! - **Log**: Append-only log file I/O
|
||||||
|
//! - **Store**: Persistent KV state from log replay
|
||||||
|
|
||||||
pub mod node;
|
pub mod node;
|
||||||
pub mod sigchain;
|
pub mod sigchain;
|
||||||
pub mod entry;
|
pub mod entry;
|
||||||
pub mod vector_clock;
|
pub mod sync_state;
|
||||||
pub mod hlc;
|
pub mod hlc;
|
||||||
pub mod clock;
|
pub mod clock;
|
||||||
pub mod proto;
|
pub mod proto;
|
||||||
pub mod data_dir;
|
pub mod data_dir;
|
||||||
pub mod signed_entry;
|
pub mod signed_entry;
|
||||||
pub mod log;
|
pub mod log;
|
||||||
|
pub mod store;
|
||||||
|
pub mod meta_store;
|
||||||
|
|
||||||
// Constants
|
// Constants
|
||||||
/// Maximum size of a serialized SignedEntry (16 MB)
|
/// Maximum size of a serialized SignedEntry (16 MB)
|
||||||
@@ -30,9 +33,13 @@ pub const MAX_ENTRY_SIZE: usize = 16 * 1024 * 1024;
|
|||||||
pub use node::Node;
|
pub use node::Node;
|
||||||
pub use sigchain::SigChain;
|
pub use sigchain::SigChain;
|
||||||
pub use entry::Entry;
|
pub use entry::Entry;
|
||||||
pub use vector_clock::VectorClock;
|
pub use sync_state::{SyncState, AuthorInfo, MissingRange};
|
||||||
pub use hlc::HLC;
|
pub use hlc::HLC;
|
||||||
pub use clock::{Clock, SystemClock, MockClock};
|
pub use clock::{Clock, SystemClock, MockClock};
|
||||||
pub use data_dir::DataDir;
|
pub use data_dir::DataDir;
|
||||||
pub use signed_entry::{EntryBuilder, sign_entry, verify_signed_entry, hash_signed_entry};
|
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;
|
||||||
|
|||||||
@@ -0,0 +1,154 @@
|
|||||||
|
//! 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";
|
||||||
|
|
||||||
|
#[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(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[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);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -28,6 +28,7 @@ pub enum NodeError {
|
|||||||
///
|
///
|
||||||
/// Each node has an Ed25519 keypair used for signing sigchain entries
|
/// Each node has an Ed25519 keypair used for signing sigchain entries
|
||||||
/// and establishing trust within the network.
|
/// and establishing trust within the network.
|
||||||
|
#[derive(Clone)]
|
||||||
pub struct Node {
|
pub struct Node {
|
||||||
signing_key: SigningKey,
|
signing_key: SigningKey,
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -31,7 +31,9 @@ mod tests {
|
|||||||
fn test_entry_with_ops() {
|
fn test_entry_with_ops() {
|
||||||
let entry = Entry {
|
let entry = Entry {
|
||||||
version: 1,
|
version: 1,
|
||||||
|
store_id: vec![1u8; 16],
|
||||||
prev_hash: vec![0u8; 32],
|
prev_hash: vec![0u8; 32],
|
||||||
|
parent_hashes: vec![],
|
||||||
seq: 5,
|
seq: 5,
|
||||||
timestamp: Some(Hlc {
|
timestamp: Some(Hlc {
|
||||||
wall_time: 1000,
|
wall_time: 1000,
|
||||||
@@ -40,7 +42,7 @@ mod tests {
|
|||||||
ops: vec![
|
ops: vec![
|
||||||
Operation {
|
Operation {
|
||||||
op_type: Some(operation::OpType::Put(PutOp {
|
op_type: Some(operation::OpType::Put(PutOp {
|
||||||
key: "/nodes/abc".to_string(),
|
key: b"/nodes/abc".to_vec(),
|
||||||
value: b"hello".to_vec(),
|
value: b"hello".to_vec(),
|
||||||
})),
|
})),
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -23,6 +23,9 @@ pub enum SigChainError {
|
|||||||
#[error("Wrong author: expected {expected}, got {got}")]
|
#[error("Wrong author: expected {expected}, got {got}")]
|
||||||
WrongAuthor { expected: String, got: String },
|
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}")]
|
#[error("Invalid sequence: expected {expected}, got {got}")]
|
||||||
InvalidSequence { expected: u64, got: u64 },
|
InvalidSequence { expected: u64, got: u64 },
|
||||||
|
|
||||||
@@ -34,11 +37,14 @@ pub enum SigChainError {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// An append-only log where each entry is cryptographically signed
|
/// 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 {
|
pub struct SigChain {
|
||||||
/// Path to the log file
|
/// Path to the log file
|
||||||
log_path: PathBuf,
|
log_path: PathBuf,
|
||||||
|
|
||||||
|
/// Store UUID (16 bytes)
|
||||||
|
store_id: [u8; 16],
|
||||||
|
|
||||||
/// Author's public key (32 bytes)
|
/// Author's public key (32 bytes)
|
||||||
author_id: [u8; 32],
|
author_id: [u8; 32],
|
||||||
|
|
||||||
@@ -50,10 +56,11 @@ pub struct SigChain {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl SigChain {
|
impl SigChain {
|
||||||
/// Create a new empty sigchain for an author
|
/// Create a new empty sigchain for a (store, author) pair
|
||||||
pub fn new(log_path: impl AsRef<Path>, author_id: [u8; 32]) -> Self {
|
pub fn new(log_path: impl AsRef<Path>, store_id: [u8; 16], author_id: [u8; 32]) -> Self {
|
||||||
Self {
|
Self {
|
||||||
log_path: log_path.as_ref().to_path_buf(),
|
log_path: log_path.as_ref().to_path_buf(),
|
||||||
|
store_id,
|
||||||
author_id,
|
author_id,
|
||||||
next_seq: 1,
|
next_seq: 1,
|
||||||
last_hash: [0u8; 32],
|
last_hash: [0u8; 32],
|
||||||
@@ -61,11 +68,11 @@ impl SigChain {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Load a sigchain from an existing log file
|
/// 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 log_path = log_path.as_ref().to_path_buf();
|
||||||
let entries = read_entries(&log_path)?;
|
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 {
|
for signed_entry in entries {
|
||||||
// Verify signature
|
// Verify signature
|
||||||
@@ -85,6 +92,18 @@ impl SigChain {
|
|||||||
// Decode Entry
|
// Decode Entry
|
||||||
let entry = Entry::decode(&signed_entry.entry_bytes[..])?;
|
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
|
// Validate sequence
|
||||||
if entry.seq != chain.next_seq {
|
if entry.seq != chain.next_seq {
|
||||||
return Err(SigChainError::InvalidSequence {
|
return Err(SigChainError::InvalidSequence {
|
||||||
@@ -156,6 +175,18 @@ impl SigChain {
|
|||||||
// Decode entry
|
// Decode entry
|
||||||
let entry = Entry::decode(&signed_entry.entry_bytes[..])?;
|
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
|
// Validate sequence
|
||||||
if entry.seq != self.next_seq {
|
if entry.seq != self.next_seq {
|
||||||
return Err(SigChainError::InvalidSequence {
|
return Err(SigChainError::InvalidSequence {
|
||||||
@@ -201,6 +232,7 @@ impl SigChain {
|
|||||||
let hlc = HLC::now_with_clock(&SystemClock);
|
let hlc = HLC::now_with_clock(&SystemClock);
|
||||||
|
|
||||||
let mut builder = EntryBuilder::new(self.next_seq, hlc)
|
let mut builder = EntryBuilder::new(self.next_seq, hlc)
|
||||||
|
.store_id(self.store_id.to_vec())
|
||||||
.prev_hash(self.last_hash.to_vec());
|
.prev_hash(self.last_hash.to_vec());
|
||||||
|
|
||||||
// Add operations
|
// Add operations
|
||||||
@@ -230,12 +262,14 @@ mod tests {
|
|||||||
temp_dir().join(format!("lattice_sigchain_test_{}.log", name))
|
temp_dir().join(format!("lattice_sigchain_test_{}.log", name))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const TEST_STORE: [u8; 16] = [1u8; 16];
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_new_sigchain() {
|
fn test_new_sigchain() {
|
||||||
let path = temp_log_path("new");
|
let path = temp_log_path("new");
|
||||||
let author = [1u8; 32];
|
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.author_id(), &author);
|
||||||
assert_eq!(chain.next_seq(), 1);
|
assert_eq!(chain.next_seq(), 1);
|
||||||
@@ -251,10 +285,11 @@ mod tests {
|
|||||||
|
|
||||||
let node = Node::generate();
|
let node = Node::generate();
|
||||||
let author = node.public_key_bytes();
|
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 clock = MockClock::new(1000);
|
||||||
let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
|
let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
|
||||||
|
.store_id(TEST_STORE.to_vec())
|
||||||
.prev_hash([0u8; 32].to_vec())
|
.prev_hash([0u8; 32].to_vec())
|
||||||
.put("/key", b"value".to_vec())
|
.put("/key", b"value".to_vec())
|
||||||
.sign(&node);
|
.sign(&node);
|
||||||
@@ -275,11 +310,12 @@ mod tests {
|
|||||||
|
|
||||||
let node = Node::generate();
|
let node = Node::generate();
|
||||||
let author = node.public_key_bytes();
|
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 clock = MockClock::new(1000);
|
||||||
|
|
||||||
for i in 1..=3 {
|
for i in 1..=3 {
|
||||||
let entry = EntryBuilder::new(i, HLC::now_with_clock(&clock))
|
let entry = EntryBuilder::new(i, HLC::now_with_clock(&clock))
|
||||||
|
.store_id(TEST_STORE.to_vec())
|
||||||
.prev_hash(chain.last_hash.to_vec())
|
.prev_hash(chain.last_hash.to_vec())
|
||||||
.put(format!("/key/{}", i), format!("value{}", i).into_bytes())
|
.put(format!("/key/{}", i), format!("value{}", i).into_bytes())
|
||||||
.sign(&node);
|
.sign(&node);
|
||||||
@@ -303,9 +339,10 @@ mod tests {
|
|||||||
|
|
||||||
// Write some entries
|
// Write some entries
|
||||||
{
|
{
|
||||||
let mut chain = SigChain::new(&path, author);
|
let mut chain = SigChain::new(&path, TEST_STORE, author);
|
||||||
for i in 1..=3 {
|
for i in 1..=3 {
|
||||||
let entry = EntryBuilder::new(i, HLC::now_with_clock(&clock))
|
let entry = EntryBuilder::new(i, HLC::now_with_clock(&clock))
|
||||||
|
.store_id(TEST_STORE.to_vec())
|
||||||
.prev_hash(chain.last_hash.to_vec())
|
.prev_hash(chain.last_hash.to_vec())
|
||||||
.put("/key", b"val".to_vec())
|
.put("/key", b"val".to_vec())
|
||||||
.sign(&node);
|
.sign(&node);
|
||||||
@@ -314,7 +351,7 @@ mod tests {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Reload from log
|
// 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.len(), 3);
|
||||||
assert_eq!(chain.next_seq(), 4);
|
assert_eq!(chain.next_seq(), 4);
|
||||||
@@ -329,11 +366,12 @@ mod tests {
|
|||||||
|
|
||||||
let node = Node::generate();
|
let node = Node::generate();
|
||||||
let author = node.public_key_bytes();
|
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 clock = MockClock::new(1000);
|
||||||
|
|
||||||
// Try to append with wrong seq (2 instead of 1)
|
// Try to append with wrong seq (2 instead of 1)
|
||||||
let entry = EntryBuilder::new(2, HLC::now_with_clock(&clock))
|
let entry = EntryBuilder::new(2, HLC::now_with_clock(&clock))
|
||||||
|
.store_id(TEST_STORE.to_vec())
|
||||||
.prev_hash([0u8; 32].to_vec())
|
.prev_hash([0u8; 32].to_vec())
|
||||||
.put("/key", b"val".to_vec())
|
.put("/key", b"val".to_vec())
|
||||||
.sign(&node);
|
.sign(&node);
|
||||||
@@ -352,11 +390,12 @@ mod tests {
|
|||||||
|
|
||||||
let node = Node::generate();
|
let node = Node::generate();
|
||||||
let author = node.public_key_bytes();
|
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 clock = MockClock::new(1000);
|
||||||
|
|
||||||
// First entry
|
// First entry
|
||||||
let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock))
|
let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock))
|
||||||
|
.store_id(TEST_STORE.to_vec())
|
||||||
.prev_hash([0u8; 32].to_vec())
|
.prev_hash([0u8; 32].to_vec())
|
||||||
.put("/key", b"v1".to_vec())
|
.put("/key", b"v1".to_vec())
|
||||||
.sign(&node);
|
.sign(&node);
|
||||||
@@ -364,6 +403,7 @@ mod tests {
|
|||||||
|
|
||||||
// Second entry with wrong prev_hash
|
// Second entry with wrong prev_hash
|
||||||
let entry2 = EntryBuilder::new(2, HLC::now_with_clock(&clock))
|
let entry2 = EntryBuilder::new(2, HLC::now_with_clock(&clock))
|
||||||
|
.store_id(TEST_STORE.to_vec())
|
||||||
.prev_hash([99u8; 32].to_vec()) // Wrong!
|
.prev_hash([99u8; 32].to_vec()) // Wrong!
|
||||||
.put("/key", b"v2".to_vec())
|
.put("/key", b"v2".to_vec())
|
||||||
.sign(&node);
|
.sign(&node);
|
||||||
@@ -382,11 +422,12 @@ mod tests {
|
|||||||
|
|
||||||
let node = Node::generate();
|
let node = Node::generate();
|
||||||
let other_author = [99u8; 32]; // Different author
|
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);
|
let clock = MockClock::new(1000);
|
||||||
|
|
||||||
// Entry signed by node but chain expects other_author
|
// Entry signed by node but chain expects other_author
|
||||||
let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
|
let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
|
||||||
|
.store_id(TEST_STORE.to_vec())
|
||||||
.prev_hash([0u8; 32].to_vec())
|
.prev_hash([0u8; 32].to_vec())
|
||||||
.put("/key", b"val".to_vec())
|
.put("/key", b"val".to_vec())
|
||||||
.sign(&node);
|
.sign(&node);
|
||||||
@@ -405,12 +446,12 @@ mod tests {
|
|||||||
|
|
||||||
let node = Node::generate();
|
let node = Node::generate();
|
||||||
let author = node.public_key_bytes();
|
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![
|
let ops = vec![
|
||||||
Operation {
|
Operation {
|
||||||
op_type: Some(operation::OpType::Put(PutOp {
|
op_type: Some(operation::OpType::Put(PutOp {
|
||||||
key: "/test".to_string(),
|
key: b"/test".to_vec(),
|
||||||
value: b"hello".to_vec(),
|
value: b"hello".to_vec(),
|
||||||
})),
|
})),
|
||||||
},
|
},
|
||||||
@@ -427,4 +468,37 @@ mod tests {
|
|||||||
|
|
||||||
std::fs::remove_file(&path).ok();
|
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 = Node::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,7 +32,9 @@ pub enum EntryError {
|
|||||||
/// Builder for creating Entry messages
|
/// Builder for creating Entry messages
|
||||||
pub struct EntryBuilder {
|
pub struct EntryBuilder {
|
||||||
version: u32,
|
version: u32,
|
||||||
|
store_id: Vec<u8>,
|
||||||
prev_hash: Vec<u8>,
|
prev_hash: Vec<u8>,
|
||||||
|
parent_hashes: Vec<Vec<u8>>,
|
||||||
seq: u64,
|
seq: u64,
|
||||||
timestamp: HLC,
|
timestamp: HLC,
|
||||||
ops: Vec<Operation>,
|
ops: Vec<Operation>,
|
||||||
@@ -43,21 +45,35 @@ impl EntryBuilder {
|
|||||||
pub fn new(seq: u64, timestamp: HLC) -> Self {
|
pub fn new(seq: u64, timestamp: HLC) -> Self {
|
||||||
Self {
|
Self {
|
||||||
version: 1,
|
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,
|
seq,
|
||||||
timestamp,
|
timestamp,
|
||||||
ops: Vec::new(),
|
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 {
|
pub fn prev_hash(mut self, hash: impl Into<Vec<u8>>) -> Self {
|
||||||
self.prev_hash = hash.into();
|
self.prev_hash = hash.into();
|
||||||
self
|
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
|
/// 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 {
|
self.ops.push(Operation {
|
||||||
op_type: Some(operation::OpType::Put(PutOp {
|
op_type: Some(operation::OpType::Put(PutOp {
|
||||||
key: key.into(),
|
key: key.into(),
|
||||||
@@ -68,7 +84,7 @@ impl EntryBuilder {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Add a Delete operation
|
/// 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 {
|
self.ops.push(Operation {
|
||||||
op_type: Some(operation::OpType::Delete(DeleteOp {
|
op_type: Some(operation::OpType::Delete(DeleteOp {
|
||||||
key: key.into(),
|
key: key.into(),
|
||||||
@@ -87,7 +103,9 @@ impl EntryBuilder {
|
|||||||
pub fn build(self) -> Entry {
|
pub fn build(self) -> Entry {
|
||||||
Entry {
|
Entry {
|
||||||
version: self.version,
|
version: self.version,
|
||||||
|
store_id: self.store_id,
|
||||||
prev_hash: self.prev_hash,
|
prev_hash: self.prev_hash,
|
||||||
|
parent_hashes: self.parent_hashes,
|
||||||
seq: self.seq,
|
seq: self.seq,
|
||||||
timestamp: Some(Hlc {
|
timestamp: Some(Hlc {
|
||||||
wall_time: self.timestamp.wall_time,
|
wall_time: self.timestamp.wall_time,
|
||||||
|
|||||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,179 @@
|
|||||||
|
//! Sync state for causality tracking and reconciliation
|
||||||
|
|
||||||
|
use std::collections::HashMap;
|
||||||
|
|
||||||
|
/// Author ID type (32-byte Ed25519 public key)
|
||||||
|
pub type Author = [u8; 32];
|
||||||
|
|
||||||
|
/// Per-author sync information (seq + hash for resume).
|
||||||
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||||
|
pub struct AuthorInfo {
|
||||||
|
pub seq: u64,
|
||||||
|
pub hash: [u8; 32],
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Sync state tracking per-author sequence numbers and hashes.
|
||||||
|
///
|
||||||
|
/// Used during reconciliation to identify missing entries between peers.
|
||||||
|
/// Each author's highest seen sequence number and hash is tracked.
|
||||||
|
#[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
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Set the info for an author.
|
||||||
|
pub fn set(&mut self, author: Author, seq: u64, hash: [u8; 32]) {
|
||||||
|
self.authors.insert(author, AuthorInfo { 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.
|
||||||
|
/// Each range includes the hash to resume reading after.
|
||||||
|
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);
|
||||||
|
if peer_info.seq > my_seq {
|
||||||
|
// We need entries from my_seq+1 to peer_info.seq
|
||||||
|
// Use our hash (or zero if we have nothing) as resume point
|
||||||
|
let from_hash = self.get(author)
|
||||||
|
.map(|i| i.hash)
|
||||||
|
.unwrap_or([0u8; 32]);
|
||||||
|
|
||||||
|
missing.push(MissingRange {
|
||||||
|
author: *author,
|
||||||
|
from_seq: my_seq,
|
||||||
|
from_hash,
|
||||||
|
to_seq: peer_info.seq,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
missing
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Merge another sync state into this one (take max seq per author).
|
||||||
|
pub fn merge(&mut self, other: &SyncState) {
|
||||||
|
for (author, info) in other.authors() {
|
||||||
|
let my_seq = self.seq(author);
|
||||||
|
if info.seq > my_seq {
|
||||||
|
self.set(*author, info.seq, info.hash);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[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
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
+31
-4
@@ -20,15 +20,42 @@ message Entry {
|
|||||||
// Versioning allows us to change the format radically later if needed
|
// Versioning allows us to change the format radically later if needed
|
||||||
uint32 version = 1;
|
uint32 version = 1;
|
||||||
|
|
||||||
|
// Store this entry belongs to (16-byte UUID)
|
||||||
|
bytes store_id = 6;
|
||||||
|
|
||||||
// Ordering Metadata
|
// 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
|
uint64 seq = 3; // Monotonic sequence number
|
||||||
HLC timestamp = 4; // Hybrid Logical Clock
|
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
|
// The Batch of Operations
|
||||||
repeated Operation ops = 5;
|
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
|
// Hybrid Logical Clock
|
||||||
message HLC {
|
message HLC {
|
||||||
uint64 wall_time = 1; // Unix timestamp (ms)
|
uint64 wall_time = 1; // Unix timestamp (ms)
|
||||||
@@ -46,12 +73,12 @@ message Operation {
|
|||||||
}
|
}
|
||||||
|
|
||||||
message PutOp {
|
message PutOp {
|
||||||
string key = 1;
|
bytes key = 1;
|
||||||
bytes value = 2; // Raw bytes allows storing images, JSON, binary, etc.
|
bytes value = 2;
|
||||||
}
|
}
|
||||||
|
|
||||||
message DeleteOp {
|
message DeleteOp {
|
||||||
string key = 1;
|
bytes key = 1;
|
||||||
}
|
}
|
||||||
|
|
||||||
// 4. The Sync Handshake (Vector Clocks)
|
// 4. The Sync Handshake (Vector Clocks)
|
||||||
|
|||||||
Reference in New Issue
Block a user