From 641db25424677198fa98629e9783241d2cabf8ab Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Wed, 20 Aug 2025 12:09:41 +0100 Subject: [PATCH 1/3] TODO for review --- CLAUDE.md | 148 +++++++++++++++++++++++ TODO.md | 343 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 491 insertions(+) create mode 100644 CLAUDE.md create mode 100644 TODO.md diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..ab701ae --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,148 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Project Overview + +This is a proof-of-concept implementation of a distributed, eventually consistent key-value store using CLSet (Causal-Length Set) CRDT. The system allows multiple peers to perform concurrent operations (insertions, updates, deletions) while automatically resolving conflicts through deterministic rules. + +## Essential Commands + +### Running the System + +**Start a peer:** +```bash +go run ./cmd/srv +``` + +**Examples:** +```bash +# Start first peer +go run ./cmd/srv peer1 "" 9000 8080 + +# Start second peer (auto-discovery via mDNS) +go run ./cmd/srv peer2 "" 9001 8081 + +# Start second peer with explicit bootstrap +go run ./cmd/srv peer2 /ip4/127.0.0.1/tcp/9000/p2p/ 9001 8081 +``` + +### Testing + +**Run all tests:** +```bash +go test ./... +``` + +**Run specific test files:** +```bash +go test -v ./crdt_test.go +go test -v ./p2p_sync_test.go +``` + +### Load Testing + +**Start two peers for load testing:** +```bash +go run ./cmd/srv peer1 "" 9000 8080 & +go run ./cmd/srv peer2 "" 9001 8081 & +``` + +**Run load test:** +```bash +go run ./cmd/load run constant -r 2000 -i 10000 -d 4000s -c 4000 crdt-load +``` + +### HTTP API Usage + +**Set a key:** +```bash +curl -X PUT -H "Content-Type: application/json" -d '{"value":"hello"}' http://localhost:8080/key/mykey +``` + +**Get a key:** +```bash +curl http://localhost:8080/key/mykey +``` + +**Delete a key:** +```bash +curl -X DELETE http://localhost:8080/key/mykey +``` + +**Get key count:** +```bash +curl http://localhost:8080/count +``` + +## Architecture Overview + +### Core Components + +**Main Implementation Files:** +- `types.go` - Core data structures (`CRDTKeyMeta`, `KeyEntry`) +- `crdt.go` - CRDT logic with conflict resolution +- `p2p_sync.go` - Peer-to-peer synchronization mechanisms +- `http_api.go` - HTTP REST API server + +**Applications:** +- `cmd/srv/main.go` - Main server application for running peer nodes +- `cmd/load/` - Load testing tools using f1 framework +- `cmd/example/main.go` - Example/demonstration application + +### CLSet CRDT Implementation + +**Key Metadata Structure:** +- `CausalLength` - Tracks insertion/deletion state (odd=exists, even=deleted) +- `ValueVersion` - Counter for value updates within same causal length +- `PeerID` - Identifier of peer that made the last change +- `PeerSeq` - Sequence number from the authoring peer + +**Conflict Resolution Rules (in order of precedence):** +1. Higher causal length wins +2. If equal causal length, higher value version wins +3. If both equal, lexicographically higher value wins +4. If all equal, lexicographically higher peer ID wins + +### P2P Synchronization + +**Peer Tracking:** +- Each peer maintains a `trackedPeers` map: `peerID -> latestKnownSequence` +- Sequence numbers are per-peer and increment on each local operation +- Sync requests include requester's tracked peers map + +**Sync Protocol:** +- `GetLatestChanges(requestorTrackedPeers, requestorPeerID)` returns missing changes +- Changes are filtered to exclude operations already known by requestor +- Tracked peers maps are merged during sync to propagate knowledge + +### Storage Layer + +**Badger KV Database:** +- Keys prefixed by namespace: `data/`, `tracked/` +- Data serialized using Go's `gob` encoding +- Atomic transactions ensure consistency during updates + +### Network Layer + +**libp2p Integration:** +- Automatic peer discovery via mDNS +- Manual peer connection via multiaddr +- Custom P2P protocol for CRDT synchronization +- HTTP API for external client interaction + +## Key Dependencies + +- `github.com/dgraph-io/badger/v4` - Embedded key-value database +- `github.com/libp2p/go-libp2p` - P2P networking stack +- `github.com/gorilla/mux` - HTTP request router +- `github.com/form3tech-oss/f1/v2` - Load testing framework +- `github.com/stretchr/testify` - Testing assertions + +## Development Notes + +- This is a proof-of-concept implementation inspired by CR-SQLite +- Uses Go 1.24.4 with standard tooling (no custom linters/formatters configured) +- Load testing requires at least 2 running peers on ports 8080 and 8081 +- Peers automatically discover each other via mDNS on the same network +- All operations are eventually consistent across all connected peers \ No newline at end of file diff --git a/TODO.md b/TODO.md new file mode 100644 index 0000000..ae9de5d --- /dev/null +++ b/TODO.md @@ -0,0 +1,343 @@ +# TODO.md + +## Architectural Improvements + +### 1. Replace Direct Badger with go-datastore Interface +- [ ] Abstract storage layer using `github.com/ipfs/go-datastore` interface +- [ ] Replace `db *badger.DB` with `store ds.Datastore` +- [ ] Create datastore wrapper for existing Badger implementation +- [ ] Allow pluggable backends (memory, leveldb, badger, etc.) +- [ ] Update tests to work with datastore interface + +### 2. Replace P2P Sync with IPFS BitSwap (**REVOLUTIONARY SIMPLIFICATION**) + +**Critical Insight**: Current P2P sync is **reimplementing BitSwap poorly** +- **Current approach**: Manual "have/want" exchange via `GetLatestChanges(requestorTracked, peerID)` +- **BitSwap approach**: Proven torrent-style block exchange used by millions of IPFS nodes +- **Why change**: We're solving the same problem (efficient data exchange) but worse + +**Current Problems with Custom P2P Sync**: +1. **O(n²) scaling**: 100 peers = 9,900 sync operations every 5 seconds +2. **Bandwidth waste**: Each peer requests same changes from multiple sources +3. **Code complexity**: ~300 lines of custom sync logic that's hard to debug +4. **No multi-peer optimization**: Can't fetch different blocks from different peers +5. **Reinventing solved problems**: Network resilience, deduplication, routing + +**Solution: BitSwap + BlockStore + GossipSub Architecture** +- [ ] **Replace entire P2P sync with BitSwap exchange**: + ```go + func (c *CRDT) Set(key, value string) { + // 1. Apply change locally + change := c.applyLocalChange(key, value) + + // 2. Store change as content-addressed block + block := blocks.NewBlock(protobuf.Marshal(change)) + c.blockstore.Put(ctx, block) + + // 3. Announce block availability to BitSwap + c.bitswap.NotifyNewBlocks(ctx, block.Cid()) + + // 4. Gossip only the tiny CID (36 bytes vs 200+ bytes full change) + c.gossip.Publish("clset-changes", block.Cid().Bytes()) + } + + func (c *CRDT) onGossipMessage(cidBytes []byte) { + cid := cid.Cast(cidBytes) + // BitSwap automatically fetches block from best available peer + block, err := c.bitswap.GetBlock(ctx, cid) + if err == nil { + change := protobuf.Unmarshal(block.RawData()) + c.mergeRemoteChange(change) + } + } + ``` + +**Massive Benefits**: + +1. **Code Reduction** (-70% sync code): + - [ ] **Delete custom sync protocol** (~300 lines removed) + - [ ] **Delete tracked peers logic** (BitSwap handles this) + - [ ] **Delete GetLatestChanges complexity** (BitSwap want/have lists) + - [ ] **Delete manual deduplication** (content addressing does this) + +2. **Performance Improvements**: + - [ ] **Multi-peer fetching** - Get different blocks from different peers simultaneously + - [ ] **Automatic bandwidth optimization** - BitSwap chooses fastest peers + - [ ] **Gossip efficiency** - 36-byte CIDs vs 200+ byte full changes (83% bandwidth reduction) + - [ ] **Content deduplication** - Same change = same CID = fetched once globally + +3. **Network Resilience**: + - [ ] **Any-peer-can-serve** - Don't need specific peer, any peer with block works + - [ ] **Automatic failover** - If peer A is slow, BitSwap tries peer B automatically + - [ ] **Network-aware routing** - BitSwap optimizes for network topology + - [ ] **Partition tolerance** - Battle-tested in IPFS across network splits + +4. **Production Battle-Testing**: + - [ ] **IPFS scale proven** - Handles millions of nodes in production + - [ ] **Active maintenance** - Core IPFS team maintains BitSwap + - [ ] **Performance optimized** - Years of optimization for global scale + - [ ] **Well documented** - Extensive docs and tooling + +**Implementation Strategy**: +- [ ] **Phase 1**: Add blockstore layer, keep current sync as fallback +- [ ] **Phase 2**: Integrate BitSwap exchange, parallel operation +- [ ] **Phase 3**: Migrate gossip to CID-only broadcasts +- [ ] **Phase 4**: Remove legacy P2P sync code +- [ ] **Result**: Simpler, faster, more reliable, less code to maintain + +### 3. Enable libp2p Relay Support +- [ ] Remove `libp2p.DisableRelay()` configuration +- [ ] Add relay discovery and configuration +- [ ] Enable multi-hop connectivity for NAT traversal +- [ ] Support nodes behind firewalls/NAT +- [ ] Add relay server mode for well-connected nodes + +### 4. Implement Batching for Performance +- [ ] Add write batching to group operations +- [ ] Implement configurable batch size and timeout +- [ ] Batch sync messages to reduce network overhead +- [ ] Add metrics for batch performance +- [ ] Consider using channels for batch accumulation + +### 5. Resilient Peer Discovery (Production-Critical) +- [ ] **DHT-based peer discovery** - Essential for NAT traversal in production +- [ ] **Bootstrap nodes** - Required for multi-datacenter deployment +- [ ] **Peer exchange protocol** - Needed for automatic topology management +- [ ] **Connection health monitoring** - Critical for 100K+ ops/sec reliability +- [ ] **Automatic failover** - Non-negotiable for high availability + +### 6. Performance at Scale (Neelix requires 100K+ ops/sec) +- [ ] **Parallel sync with multiple peers** - Required for throughput +- [ ] **Connection pooling optimization** - Critical despite libp2p (HTTP connection reuse) +- [ ] **Protobuf serialization** - Smaller messages = better network efficiency +- [ ] **Sync checkpoints/resume** - Prevent full resync during network issues +- [ ] **Compression for sync messages** - Bandwidth optimization for large datasets +- [ ] **Async work queues** - Decouple API from CRDT operations (like Neelix) + +### 7. Production Monitoring (Operational Requirements) +- [ ] **Comprehensive metrics** - sync latency, conflict rates, queue sizes +- [ ] **Distributed tracing** - Essential for debugging multi-node issues +- [ ] **Structured logging** with correlation IDs across nodes +- [ ] **Health monitoring** - Node failure detection and alerting +- [ ] **Performance dashboards** - Required for 100K+ ops/sec monitoring +- [ ] **Circuit breakers** - Prevent cascade failures + +### 8. Enterprise Testing (Production Validation) +- [ ] **Chaos testing** - Network partitions, node failures, Byzantine faults +- [ ] **Load testing framework** - Must handle 100K+ concurrent operations +- [ ] **Property-based CRDT tests** - Verify invariants under all conditions +- [ ] **Multi-datacenter simulation** - Test geographic distribution +- [ ] **Fuzz testing** - Discover edge cases in sync protocol +- [ ] **Performance regression testing** - Ensure scalability doesn't degrade + +### 9. Documentation (User-Focused) +- [ ] Create production deployment guide +- [ ] Document configuration parameters +- [ ] Add troubleshooting guide for common issues +- [ ] **Skip complex ADRs** - architecture is intentionally simple + +## Features Required for Production Use (Based on Neelix Analysis) + +### 11. Event Hooks System (Essential for Reactive Applications) +- [ ] Add hooks to CRDT struct for local operations: + - [ ] `OnSet(key, value string, meta CRDTKeyMeta)` - triggered after local set + - [ ] `OnDelete(key string, meta CRDTKeyMeta)` - triggered after local delete + - [ ] `OnSync(changes []KeyEntry, fromPeer string)` - triggered after sync +- [ ] Hooks should be called **after** database persistence +- [ ] Keep hooks simple - just notification, not transformation +- [ ] Add optional hook timeout to prevent blocking + +### 12. Peer Lifecycle Management +- [ ] Track connected peers with simple metadata in memory +- [ ] Add peer state: `connecting`, `active`, `disconnecting`, `failed` +- [ ] Implement connection health monitoring (ping/heartbeat) +- [ ] Add graceful disconnect notification to peers +- [ ] Remove failed peers from sync rotation automatically +- [ ] **No complex membership DAG** - use simple peer list + +### 13. Query Operations (Leverage existing CLSet structure) +- [ ] Add `ListKeys(prefix string) []string` using existing iteration +- [ ] Add `ListEntries(prefix string) []KeyEntry` for full data +- [ ] Add pagination with `offset` and `limit` parameters +- [ ] Use existing key count logic but make it prefix-aware +- [ ] **No complex query engine** - simple prefix matching + +### 14. Configuration Structure (CLSet-specific) +- [ ] Create focused Options for clset-example: + ```go + type Config struct { + SyncInterval time.Duration // How often to sync + SyncBatchSize int // Max changes per sync + PeerTimeout time.Duration // When to consider peer failed + MaxPeers int // Connection limit + EnableHooks bool // Enable/disable hooks + } + ``` +- [ ] Focus on sync and peer management, not DAG/IPLD concepts +- [ ] Add validation specific to CLSet constraints + +### 15. Clean Shutdown (Simple notification) +- [ ] Send "shutting down" message to connected peers +- [ ] Finish pending syncs with short timeout +- [ ] Close database cleanly +- [ ] **No complex acknowledgment protocol** - keep it simple + +### 16. Compaction (Distributed Consensus Challenge) +**Problem**: Change entries (`change/peerID/seq`) accumulate indefinitely. Safe compaction requires ensuring no peer needs deleted changes for sync. + +**Solution: Multi-Policy Conservative Compaction** + +#### Core Challenge: Split-Brain Safe Compaction +- Unlike go-ds-crdt's Merkle-DAG consensus, CLSet needs distributed agreement on what's safe to delete +- Must handle network partitions where different peer groups compact independently +- Need to balance storage growth vs safety guarantees + +#### Proposed Implementation: +- [ ] **Extended Peer Knowledge Tracking**: + ```go + type CRDT struct { + trackedPeers map[string]uint64 // What I know about each peer + peerKnowledge map[string]map[string]uint64 // What each peer knows about others + lastPeerSync map[string]time.Time // When I last synced with each peer + } + ``` + +- [ ] **Conservative Compaction Policy**: + ```go + type CompactionConfig struct { + MinAge time.Duration // Never compact changes newer than this + MinRetainPerPeer int // Always keep N recent changes per peer + RequireAckFrom []string // Must have ack from these "anchor" peers + PeerTimeoutThreshold time.Duration // When to consider peer permanently gone + CompactionInterval time.Duration // How often to attempt compaction + MaxBatchSize int // Limit compaction batch size + } + ``` + +- [ ] **Three-Level Safety Check**: + 1. **Time-based**: Never compact changes newer than `MinAge` + 2. **Count-based**: Always retain `MinRetainPerPeer` recent changes + 3. **Peer-based**: Only compact sequences acknowledged by connected peers + +- [ ] **Safe Compaction Algorithm**: + ```go + func (c *CRDT) SafeCompactionPoint(peerID string) uint64 { + // 1. Time cutoff - conservative baseline + timeCutoff := time.Now().Add(-c.config.MinAge) + timeBasedSeq := c.getSequenceAtTime(peerID, timeCutoff) + + // 2. Count cutoff - operational minimum + totalChanges := c.countChanges(peerID) + countBasedSeq := max(0, totalChanges - c.config.MinRetainPerPeer) + + // 3. Peer acknowledgment cutoff + peerBasedSeq := c.getMinAcknowledgedSequence(peerID) + + // Use most conservative (smallest) value + return min(timeBasedSeq, countBasedSeq, peerBasedSeq) + } + ``` + +- [ ] **Gradual Compaction**: + - Delete changes in small batches to avoid blocking operations + - Use background goroutine with configurable interval + - Add compaction metrics (sequences compacted, storage freed) + +- [ ] **Split-Brain Protection**: + - Require minimum number of "anchor" peers for aggressive compaction + - Fall back to time-based only during network partitions + - Add operator override for emergency compaction + +#### Advantages over Merkle-DAG Compaction: +- **Simpler logic**: Just delete old `change/peerID/*` entries, no DAG merging +- **Per-peer isolation**: Can compact different peers independently +- **Predictable storage**: Upper bound = `NumPeers × MinRetainPerPeer × AvgChangeSize` +- **Operational control**: Policies configurable by operators, not consensus algorithm + +#### Trade-offs: +- More conservative than optimal (keeps more data than strictly necessary) +- Requires careful tuning of retention policies +- Still needs peer connectivity for optimal compaction + +### 17. Sync Improvements (Build on existing protocol) +- [ ] Add chunked sync for large change sets (use existing 100k limit) +- [ ] Implement sync resume after connection failure +- [ ] Add change checksums for sync verification +- [ ] **Keep simple request/response model** - no complex DAG walking + +### 18. Operational Features +- [ ] Add basic metrics: sync count, peer count, operation latency +- [ ] Add structured logging with configurable levels +- [ ] Add health check endpoint for load balancers +- [ ] **Focus on CLSet-specific metrics** - causal length conflicts, etc. + +## Priority Order (Production Distributed Systems) + +### Phase 1: Core Distributed Systems Infrastructure +1. **BitSwap + BlockStore integration** (#2) - **REVOLUTIONARY CHANGE** - Replace custom P2P with battle-tested IPFS exchange +2. **Datastore abstraction** (#1) - Foundation for scalability +3. **Event hooks system** (#11) - **CRITICAL** - Required for reactive applications like Neelix +4. **Resilient peer discovery** (#5) - **CRITICAL** - DHT + bootstrap for production deployment +5. **libp2p relay support** (#3) - **CRITICAL** - NAT traversal for real networks + +### Phase 2: Performance at Scale (100K+ ops/sec requirements) +6. **Performance optimizations** (#6) - **CRITICAL** - Parallel sync, connection pooling, compression +7. **Batching implementation** (#4) - **CRITICAL** - Required for throughput +8. **Peer lifecycle management** (#12) - **CRITICAL** - Health monitoring, automatic failover +9. **Query operations** (#13) - **CRITICAL** - Needed for state inspection at scale +10. **Compaction** (#16) - **CRITICAL** - Storage management with distributed consensus + +### Phase 3: Production Reliability +11. **Production monitoring** (#7) - **CRITICAL** - Comprehensive metrics, tracing, alerting +12. **Configuration system** (#14) - **CRITICAL** - Operational flexibility and tuning +13. **Sync improvements** (#17) - **CRITICAL** - Checkpoints, resume, verification +14. **Clean shutdown** (#15) - **CRITICAL** - Graceful degradation +15. **Operational features** (#18) - **CRITICAL** - Circuit breakers, backpressure + +### Phase 4: Production Validation & Operations +16. **Enterprise testing** (#8) - **CRITICAL** - Chaos, load, property-based testing +17. **Documentation** (#9) - **CRITICAL** - Production deployment guides + +## Implementation Strategy (Revised) + +### Goal: Drop-in Replacement for go-ds-crdt in Neelix +- Must handle **100K+ concurrent operations** +- Must support **multi-datacenter deployment** +- Must provide **partition tolerance** and **high availability** +- Must maintain **CLSet conflict resolution advantages** while scaling + +### Success Criteria: +- [ ] Neelix can replace go-ds-crdt dependency with zero application changes +- [ ] Performance equals or exceeds go-ds-crdt at 100K+ ops/sec +- [ ] Network partition tolerance demonstrated through chaos testing +- [ ] Multi-node deployment validated in production-like environment + +## Implementation Strategy + +### Step 1: Create Compatibility Layer +- Build a go-ds-crdt compatible API on top of clset-example +- Maintain CLSet semantics while exposing CRDT interface +- Focus on drop-in replacement capability + +### Step 2: Incremental Feature Addition +- Add features in priority order +- Maintain backward compatibility +- Keep simple mode available + +### Step 3: Performance Validation +- Benchmark against go-ds-crdt +- Test with Neelix-scale workloads (distributed IP allocation) +- Optimize critical paths + +### Step 4: Migration Tools +- Create migration guide from go-ds-crdt +- Build compatibility test suite +- Provide migration utilities + +## Notes + +- Neelix demonstrates sophisticated CRDT usage in production +- Event hooks and membership management are critical for real applications +- Must support both simple and complex deployment scenarios +- Performance must scale to thousands of operations per second +- Network partition tolerance is non-negotiable \ No newline at end of file From 0ab13db2590269f6073b5c7f19a2d831fc0e462d Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Wed, 20 Aug 2025 12:40:35 +0100 Subject: [PATCH 2/3] Update --- TODO.md | 204 ++++++++++++++++++-------------------------------------- 1 file changed, 65 insertions(+), 139 deletions(-) diff --git a/TODO.md b/TODO.md index ae9de5d..f012d4d 100644 --- a/TODO.md +++ b/TODO.md @@ -9,81 +9,73 @@ - [ ] Allow pluggable backends (memory, leveldb, badger, etc.) - [ ] Update tests to work with datastore interface -### 2. Replace P2P Sync with IPFS BitSwap (**REVOLUTIONARY SIMPLIFICATION**) - -**Critical Insight**: Current P2P sync is **reimplementing BitSwap poorly** -- **Current approach**: Manual "have/want" exchange via `GetLatestChanges(requestorTracked, peerID)` -- **BitSwap approach**: Proven torrent-style block exchange used by millions of IPFS nodes -- **Why change**: We're solving the same problem (efficient data exchange) but worse - -**Current Problems with Custom P2P Sync**: -1. **O(n²) scaling**: 100 peers = 9,900 sync operations every 5 seconds -2. **Bandwidth waste**: Each peer requests same changes from multiple sources -3. **Code complexity**: ~300 lines of custom sync logic that's hard to debug -4. **No multi-peer optimization**: Can't fetch different blocks from different peers -5. **Reinventing solved problems**: Network resilience, deduplication, routing - -**Solution: BitSwap + BlockStore + GossipSub Architecture** -- [ ] **Replace entire P2P sync with BitSwap exchange**: +### 2. Optimize P2P Sync with Sequence Number Gossip (**SMART OPTIMIZATION**) + +**Current Analysis** (corrected based on review): +- **Current approach works correctly** - maintains CLSet ordering requirements +- **Change cleanup exists** - superseded entries are deleted automatically +- **O(n²) is the main scalability issue** - 100 peers = 9,900 sync operations every 5 seconds + +**Problem**: Full sync requests even when peers have minimal differences + +**Solution: Sequence Number Gossip + Selective Sync** +- [ ] **Gossip peer sequence summaries** instead of full sync requests: ```go - func (c *CRDT) Set(key, value string) { - // 1. Apply change locally - change := c.applyLocalChange(key, value) - - // 2. Store change as content-addressed block - block := blocks.NewBlock(protobuf.Marshal(change)) - c.blockstore.Put(ctx, block) - - // 3. Announce block availability to BitSwap - c.bitswap.NotifyNewBlocks(ctx, block.Cid()) - - // 4. Gossip only the tiny CID (36 bytes vs 200+ bytes full change) - c.gossip.Publish("clset-changes", block.Cid().Bytes()) + type PeerSummary struct { + PeerID string // Who is reporting + Sequences map[string]uint64 // peerID -> latest sequence I have + Timestamp time.Time // When this summary was created } - func (c *CRDT) onGossipMessage(cidBytes []byte) { - cid := cid.Cast(cidBytes) - // BitSwap automatically fetches block from best available peer - block, err := c.bitswap.GetBlock(ctx, cid) - if err == nil { - change := protobuf.Unmarshal(block.RawData()) - c.mergeRemoteChange(change) + func (c *CRDT) gossipSummary() { + summary := PeerSummary{ + PeerID: c.PeerID, + Sequences: c.GetTrackedPeers(), // What I have from each peer + Timestamp: time.Now(), } + c.gossip.Publish("peer-summaries", summary) } ``` -**Massive Benefits**: +- [ ] **Selective sync based on sequence gaps**: + ```go + func (c *CRDT) onSummaryMessage(summary PeerSummary) { + // Find peers that have sequences I don't have + for peerID, theirSeq := range summary.Sequences { + mySeq := c.trackedPeers[peerID] + if theirSeq > mySeq { + // They have changes I don't - fetch from them specifically + go c.syncSpecificPeer(summary.PeerID, peerID, mySeq+1, theirSeq) + } + } + } + ``` -1. **Code Reduction** (-70% sync code): - - [ ] **Delete custom sync protocol** (~300 lines removed) - - [ ] **Delete tracked peers logic** (BitSwap handles this) - - [ ] **Delete GetLatestChanges complexity** (BitSwap want/have lists) - - [ ] **Delete manual deduplication** (content addressing does this) +**Benefits of This Approach**: -2. **Performance Improvements**: - - [ ] **Multi-peer fetching** - Get different blocks from different peers simultaneously - - [ ] **Automatic bandwidth optimization** - BitSwap chooses fastest peers - - [ ] **Gossip efficiency** - 36-byte CIDs vs 200+ byte full changes (83% bandwidth reduction) - - [ ] **Content deduplication** - Same change = same CID = fetched once globally +1. **Maintains CLSet Correctness**: + - [ ] **Preserves ordering** - sequences fetched in correct order + - [ ] **No delivery race conditions** - direct P2P maintains causality + - [ ] **Deterministic conflict resolution** - CLSet semantics unchanged -3. **Network Resilience**: - - [ ] **Any-peer-can-serve** - Don't need specific peer, any peer with block works - - [ ] **Automatic failover** - If peer A is slow, BitSwap tries peer B automatically - - [ ] **Network-aware routing** - BitSwap optimizes for network topology - - [ ] **Partition tolerance** - Battle-tested in IPFS across network splits +2. **Dramatic Efficiency Gains**: + - [ ] **Gossip tiny summaries** (~200 bytes) instead of requesting full syncs + - [ ] **Selective fetching** - only sync from peers with newer data + - [ ] **Reduced sync frequency** - avoid unnecessary sync requests when peers are up-to-date + - [ ] **Smart peer selection** - choose peers with most missing sequences -4. **Production Battle-Testing**: - - [ ] **IPFS scale proven** - Handles millions of nodes in production - - [ ] **Active maintenance** - Core IPFS team maintains BitSwap - - [ ] **Performance optimized** - Years of optimization for global scale - - [ ] **Well documented** - Extensive docs and tooling +3. **Preserves Current Architecture**: + - [ ] **Keep existing sync protocol** - proven to work correctly + - [ ] **Keep tracked peers logic** - essential for CLSet operation + - [ ] **Keep sequence ordering** - critical for correctness + - [ ] **Add optimization layer** - gossip summaries on top **Implementation Strategy**: -- [ ] **Phase 1**: Add blockstore layer, keep current sync as fallback -- [ ] **Phase 2**: Integrate BitSwap exchange, parallel operation -- [ ] **Phase 3**: Migrate gossip to CID-only broadcasts -- [ ] **Phase 4**: Remove legacy P2P sync code -- [ ] **Result**: Simpler, faster, more reliable, less code to maintain +- [ ] **Phase 1**: Add PeerSummary gossip alongside current sync +- [ ] **Phase 2**: Implement selective sync based on sequence gaps +- [ ] **Phase 3**: Reduce periodic full sync frequency (rely more on summaries) +- [ ] **Phase 4**: Add metrics to measure sync efficiency improvements +- [ ] **Result**: Maintain correctness while dramatically reducing sync overhead ### 3. Enable libp2p Relay Support - [ ] Remove `libp2p.DisableRelay()` configuration @@ -182,82 +174,16 @@ - [ ] Close database cleanly - [ ] **No complex acknowledgment protocol** - keep it simple -### 16. Compaction (Distributed Consensus Challenge) -**Problem**: Change entries (`change/peerID/seq`) accumulate indefinitely. Safe compaction requires ensuring no peer needs deleted changes for sync. - -**Solution: Multi-Policy Conservative Compaction** - -#### Core Challenge: Split-Brain Safe Compaction -- Unlike go-ds-crdt's Merkle-DAG consensus, CLSet needs distributed agreement on what's safe to delete -- Must handle network partitions where different peer groups compact independently -- Need to balance storage growth vs safety guarantees - -#### Proposed Implementation: -- [ ] **Extended Peer Knowledge Tracking**: - ```go - type CRDT struct { - trackedPeers map[string]uint64 // What I know about each peer - peerKnowledge map[string]map[string]uint64 // What each peer knows about others - lastPeerSync map[string]time.Time // When I last synced with each peer - } - ``` - -- [ ] **Conservative Compaction Policy**: - ```go - type CompactionConfig struct { - MinAge time.Duration // Never compact changes newer than this - MinRetainPerPeer int // Always keep N recent changes per peer - RequireAckFrom []string // Must have ack from these "anchor" peers - PeerTimeoutThreshold time.Duration // When to consider peer permanently gone - CompactionInterval time.Duration // How often to attempt compaction - MaxBatchSize int // Limit compaction batch size - } - ``` - -- [ ] **Three-Level Safety Check**: - 1. **Time-based**: Never compact changes newer than `MinAge` - 2. **Count-based**: Always retain `MinRetainPerPeer` recent changes - 3. **Peer-based**: Only compact sequences acknowledged by connected peers - -- [ ] **Safe Compaction Algorithm**: - ```go - func (c *CRDT) SafeCompactionPoint(peerID string) uint64 { - // 1. Time cutoff - conservative baseline - timeCutoff := time.Now().Add(-c.config.MinAge) - timeBasedSeq := c.getSequenceAtTime(peerID, timeCutoff) - - // 2. Count cutoff - operational minimum - totalChanges := c.countChanges(peerID) - countBasedSeq := max(0, totalChanges - c.config.MinRetainPerPeer) - - // 3. Peer acknowledgment cutoff - peerBasedSeq := c.getMinAcknowledgedSequence(peerID) - - // Use most conservative (smallest) value - return min(timeBasedSeq, countBasedSeq, peerBasedSeq) - } - ``` - -- [ ] **Gradual Compaction**: - - Delete changes in small batches to avoid blocking operations - - Use background goroutine with configurable interval - - Add compaction metrics (sequences compacted, storage freed) - -- [ ] **Split-Brain Protection**: - - Require minimum number of "anchor" peers for aggressive compaction - - Fall back to time-based only during network partitions - - Add operator override for emergency compaction - -#### Advantages over Merkle-DAG Compaction: -- **Simpler logic**: Just delete old `change/peerID/*` entries, no DAG merging -- **Per-peer isolation**: Can compact different peers independently -- **Predictable storage**: Upper bound = `NumPeers × MinRetainPerPeer × AvgChangeSize` -- **Operational control**: Policies configurable by operators, not consensus algorithm +### 16. Storage Management (Current Implementation Analysis) +**Corrected Understanding**: +- **Change entries are cleaned up automatically** - superseded entries deleted when keys updated +- **Storage is bounded** - change index has at most as many entries as KV store itself +- **No indefinite accumulation** - current implementation already handles storage management correctly -#### Trade-offs: -- More conservative than optimal (keeps more data than strictly necessary) -- Requires careful tuning of retention policies -- Still needs peer connectivity for optimal compaction +**Potential Optimizations** (if needed in future): +- [ ] **Monitor storage usage** - add metrics for change index size vs KV size ratio +- [ ] **Configurable cleanup policies** - allow operators to tune cleanup behavior +- [ ] **Storage alerts** - warn if change index grows beyond expected bounds ### 17. Sync Improvements (Build on existing protocol) - [ ] Add chunked sync for large change sets (use existing 100k limit) @@ -274,7 +200,7 @@ ## Priority Order (Production Distributed Systems) ### Phase 1: Core Distributed Systems Infrastructure -1. **BitSwap + BlockStore integration** (#2) - **REVOLUTIONARY CHANGE** - Replace custom P2P with battle-tested IPFS exchange +1. **Sequence number gossip optimization** (#2) - **SMART OPTIMIZATION** - Reduce O(n²) sync while preserving CLSet ordering 2. **Datastore abstraction** (#1) - Foundation for scalability 3. **Event hooks system** (#11) - **CRITICAL** - Required for reactive applications like Neelix 4. **Resilient peer discovery** (#5) - **CRITICAL** - DHT + bootstrap for production deployment @@ -285,7 +211,7 @@ 7. **Batching implementation** (#4) - **CRITICAL** - Required for throughput 8. **Peer lifecycle management** (#12) - **CRITICAL** - Health monitoring, automatic failover 9. **Query operations** (#13) - **CRITICAL** - Needed for state inspection at scale -10. **Compaction** (#16) - **CRITICAL** - Storage management with distributed consensus +10. **Storage monitoring** (#16) - **OPERATIONAL** - Monitor change index efficiency (already bounded) ### Phase 3: Production Reliability 11. **Production monitoring** (#7) - **CRITICAL** - Comprehensive metrics, tracing, alerting From 74cb037e5746f6b6400e594a118b18ac3da6046b Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Wed, 20 Aug 2025 12:43:32 +0100 Subject: [PATCH 3/3] Update readme --- README.md | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/README.md b/README.md index 9e2406e..253926e 100644 --- a/README.md +++ b/README.md @@ -110,6 +110,22 @@ The causal length and the value version are used as follows: The sync mechanism works as follows: * The main metadata used for syncing are the key's **peer ID** and **peer seq**. As already mentioned, whenever a peer locally creates, updates or removes a key, the key's peer ID is set to the local peer's ID, and the key's peer seq is set to the next available sequence number of the local peer. + +### Change Index Architecture + +To enable efficient synchronization, the system maintains a **change index** alongside the main key-value store. Using a SQL analogy: + +* **Main KV Store**: Like a SQL table with columns `(key, value, causal_length, value_version, peer_id, peer_seq)` +* **Change Index**: Like a secondary index on `(peer_id, peer_seq)` that allows efficient range queries + +The change index is stored using keys like `change/peerA/0000000001`, `change/peerA/0000000002`, etc. This enables the `GetLatestChanges` function to efficiently answer queries like: + +*"Give me all changes from peer X starting from sequence number Y"* + +by performing a simple key prefix scan: `change/peerX/` starting from the formatted sequence number. + +When a key is updated, the old change index entry is automatically deleted and replaced with the new one, ensuring the change index remains bounded (at most one entry per key in the KV store). + * Each peer keeps a "tracked peers" map: a mapping from peer ID to the latest sequence number "known" from remote peers. This is used to keep track of what changes they already have from all peers and is updated whenever peers merge changes from remote peers. It also contains the local peer's latest sequence number. * Each peer has a GetLatestChanges function that lets remote peers fetch changes that they don't already have. This function receives two arguments: `requestorTrackedPeers` (a copy of the tracked peers map of the requesting peer), and `requestorPeerID` (the peer ID of the requesting peer). Its return value is a pair of two items: `changes` (the list of changes that the requestor doesn't already have) and `trackedPeers` (a copy of the called peer's tracked peers map). More specifically, `changes` contains all keys from the called peer such that (1) the key's `peerID` is not equal to `requestorPeerID` and (2) either the key's `peerID` is not present in `requestorTrackedPeers` or the key's peer seq is greater than the sequence number recorded in `requestorTrackedPeers`. * A peer A can merge changes from another peer B. For that, A calls B's GetLatestChanges, then A merges the changes to its local CRDT and merges the tracked peer information provided by B into its own local tracked peers map.