diff --git a/.github/ISSUE_TEMPLATE/DMP_2026.yml b/.github/ISSUE_TEMPLATE/DMP_2026.yml new file mode 100644 index 000000000..bc6971528 --- /dev/null +++ b/.github/ISSUE_TEMPLATE/DMP_2026.yml @@ -0,0 +1,345 @@ +name: DMP 2026 Project Template +description: List a new project for Dedicated Mentoring Program (DMP) 2026 +title: "[DMP 2026]: " +labels: ["DMP 2026"] +body: + - type: textarea + id: ticket-description + validations: + required: true + attributes: + label: Ticket Contents + value: | + ## Description + [Provide a brief description of the feature, including why it is needed and what it will accomplish.] + + - type: textarea + id: ticket-goals + validations: + required: true + attributes: + label: Goals & Mid-Point Milestone + description: List the goals of the feature. Please add the goals that must be achieved by Mid-point check-in i.e 1.5 months into the coding period. + value: | + ## Goals + - [ ] [Goal 1] + - [ ] [Goal 2] + - [ ] [Goal 3] + - [ ] [Goal 4] + - [ ] [Goals Achieved By Mid-point Milestone] + + - type: textarea + id: ticket-setup + attributes: + label: Setup/Installation + description: Please list or link setup or installation guide (if any) + + - type: textarea + id: ticket-expected-outcome + attributes: + label: Expected Outcome + description: Describe in detail what the final product or result should look like and how it should behave. + + - type: textarea + id: ticket-acceptance-criteria + attributes: + label: Acceptance Criteria + description: List the acceptance criteria for this feature. + + - type: textarea + id: ticket-implementation-details + validations: + required: true + attributes: + label: Implementation Details + description: List any technical details about the proposed implementation, including any specific technologies that will be used. + + - type: textarea + id: ticket-mockups + attributes: + label: Mockups/Wireframes + description: Include links to any visual aids, mockups, wireframes, or diagrams that help illustrate what the final product should look like. This is not always necessary, but can be very helpful in many cases. + + - type: input + id: ticket-product + attributes: + label: Product Name + placeholder: Enter Product Name + validations: + required: true + + - type: dropdown + id: ticket-organisation + attributes: + label: Organisation Name + description: Enter Organisation Name + multiple: false + options: + - Agami + - Argusoft + - ARMMAN + - Avanti Fellows + - Bandhu + - Beckn + - Belongg + - Blockster Global (CREDBEL) + - Blockster Labs / AyanWorks + - CBoard + - CHAOSS + - CHAOSS Africa + GWU + - Civis + - ConveGenius + - Consul Democracy + - COSS + - CranberryFit + - Development Gateway + - DHIS2 + - Dhiway + - Dhwani + - Digital Green + - Digital India + - Dimagi + - Drupal + - Education Initiative + - eGov + - EkShop Marketplace + - FIDE + - FinternetLabs + - Flywheel + - GovDirectory + - Haqdarshak + - Healthsites.io + - IDinsight + - If Me + - IIIT Delhi + - IIT Bombay + - IIT Delhi + - Impactyaan + - Indus Action + - Intel Health + - Key Education Foundation + - Khushi Baby + - Learning Economy + - Linux Foundation + - Mecha Systems + - Medic Mobile + - Medtronic Labs + - MetaBrainz + - Mifos + - Mojaloop + - MOSIP + - NASSCOM Foundation + - NHA + - NIUA + - Norwegian Meteorological Institute + - NSUT x SEETA x AIC + - ONDC + - ONEST + - Open Healthcare Network + - OpenCRVS + - OpenFn + - OpenIMIS + - OpenMRS + - OpenSPP + - Piramal Swasthya + - Planet Read + - Policy Engine + - Pratham Books + - Project Second Chance + - Project Tech4Dev + - Protean + - RCTS-IIITH + - Reap Benefit + - Resolve to Save Lives + - Rocket Learning + - Rumsan + - Sahamati + - SamagraX + - Samanvay Foundation + - Sampatti Card + - Sanketika + - ShikshaLokam + - SimPPL + - Sugar Labs + - Swasth Alliance + - Swecha + - Tarento + - Tattle + - Tech4Dev + - Tekdi + - The Apprentice Project + - The Mifos Initiative + - Thoughtworks + - Tibil + - TinkerHub + - Trustin + - Tuner Labs + - TYCIA + - UNICEF + - United Nations + - Ushahidi + - Win Over Cancer + - WRI + - Zendalona + - Zenysis + - Arghyam + validations: + required: true + + - type: dropdown + id: ticket-governance-domain + attributes: + label: Domain + options: + - ⁠Healthcare + - ⁠Education + - Financial Inclusion + - ⁠Livelihoods + - ⁠Skilling + - ⁠Learning & Development + - ⁠Agriculture + - ⁠Service Delivery + - Open Source Library + - Water + validations: + required: true + + + - type: dropdown + id: ticket-technical-skills-required + attributes: + label: Tech Skills Needed + description: Select the technologies needed for this ticket (use Ctrl or Command to select multiple) + multiple: true + options: + - .NET + - Angular + - Artificial Intelligence + - ASP.NET + - AWS + - Babel + - Bootstrap + - C# + - Chart.js + - CI/CD + - Computer Vision + - CORS + - cURL + - Cypress + - D3.js + - Database + - Debugging + - Design + - DevOps + - Django + - Docker + - Electron + - ESLint + - Express.js + - Feature + - Flask + - Go + - GraphQL + - HTML + - Ionic + - Jest + - Java + - JavaScript + - Jenkins + - JWT + - Kubernetes + - Laravel + - Machine Learning + - Maintenance + - Markdown + - Material-UI + - Microservices + - MongoDB + - Mobile + - Mockups + - Mocha + - Natural Language Processing + - NestJS + - Node.js + - NUnit + - OAuth + - Performance Improvement + - Prettier + - Python + - Question + - React + - React Native + - Redux + - RESTful APIs + - Ruby + - Ruby on Rails + - Rust + - Scala + - Security + - Selenium + - SEO + - Serverless + - Solidity + - Spring Boot + - SQL + - Swagger + - Tailwind CSS + - Test + - Testing Library + - Three.js + - TypeScript + - UI/UX/Design + - Virtual Reality + - Vue.js + - WebSockets + - Webpack + - Other + validations: + required: true + + - type: textarea + id: ticket-mentors + attributes: + label: Mentor(s) + description: Please tag relevant mentors for the ticket + validations: + required: true + + - type: dropdown + id: ticket-category + attributes: + label: Category + description: Choose the categories that best describe your ticket + multiple: true + options: + - API + - Analytics + - Accessibility + - Backend + - Breaking Change + - Beginner Friendly + - Configuration + - CI/CD + - Database + - Data Science + - Deprecation + - Documentation + - Delpoyment + - Frontend + - Internationalization + - Localization + - Machine Learning + - Maintenance + - Mobile + - Performance Improvement + - Question + - Refactoring + - Research + - Needs Reproduction + - SEO + - Security + - Testing + - AI + - Other + validations: + required: true diff --git a/examples/p2pcalc/README.md b/examples/p2pcalc/README.md new file mode 100644 index 000000000..d4de7d68c --- /dev/null +++ b/examples/p2pcalc/README.md @@ -0,0 +1,417 @@ +# P2PCalc : Decentralised EtherCalc over py-libp2p GossipSub + +**Author:** Dashpreet Singh + +**Email:** dashpreetsinghhanda@gmail.com | 2024ucs0087@iitjammu.ac.in + +**Institution:** IIT Jammu, B.Tech CSE 2024-2028 + +**Issue:** [DMP 2026 #34](https://github.com/seetadev/py-libp2p/issues/34) + +______________________________________________________________________ + +## What this is + +P2PCalc replaces EtherCalc's centralised Redis pub-sub synchronisation layer +with py-libp2p GossipSub. Multiple peers can collaboratively edit the same +spreadsheet in real time , no central server, no single point of failure, +no cloud dependency. + +The browser-facing EtherCalc UI and the Node.js worker are completely +untouched. P2PCalc intercepts at the Redis pub-sub layer, the only place +where EtherCalc's components talk to each other. Everything else stays the +same. + +______________________________________________________________________ + +## The problem with EtherCalc's current architecture + +EtherCalc works by having a central Node.js server relay SocialCalc command +strings between browser clients over WebSockets. Every edit goes: + +``` +Browser A -> WebSocket -> Node.js server -> Redis pub-sub -> Node.js server -> WebSocket -> Browser B +``` + +This architecture has three hard problems: + +**Single point of failure.** If the server goes down, collaboration stops. +Peers with pending edits lose them. + +**No offline-first capability.** A peer that loses connectivity cannot +continue editing and reconnect later. There is no mechanism to reconcile +diverged states. + +**Centralised trust.** All edits pass through one server. In civic-tech, +disaster response, or low-connectivity rural deployments, this is a +fundamental constraint not an incidental one. + +______________________________________________________________________ + +## The approach + +The key insight from studying EtherCalc's internals is that the server +does not compute, it just relays. Every SocialCalc command is executed +client-side in the browser's JavaScript engine. The server is a glorified +message bus. + +If we can replace that message bus with a decentralised one, we get +peer-to-peer collaboration for free, with no changes to the UI or the +client-side computation engine. + +EtherCalc uses Redis pub-sub channels internally, one per room, with the +channel name `sc:`. Every command string (`set A1 value 42`, +`insertrow 3 1`, etc.) is published to this channel and consumed by the +Node.js worker which relays it to connected browsers. + +P2PCalc taps into this channel. The adapter subscribes to `sc:`, +wraps each command in a typed operation envelope, and broadcasts it via +GossipSub to all other peers on the topic `p2pcalc/`. Incoming +operations from remote peers are unwrapped and injected back into the same +Redis channel. EtherCalc's existing Socket.io layer delivers them to +browsers as if they came from a local user. + +``` +Before (centralised): + Browser -> WS -> Node.js -> Redis -> Node.js -> WS -> Browser + +After (P2PCalc): + Browser -> WS -> Node.js -> Redis -> P2PCalc Adapter + | + GossipSub topic + p2pcalc/ + | + Remote P2PCalc Adapter + | + Redis -> Node.js -> WS -> Browser +``` + +The browser never knows the difference. + +______________________________________________________________________ + +## Innovations + +### 1. Hybrid Logical Clocks instead of wall clocks + +Most collaborative editors that claim to handle concurrent edits use wall-clock +timestamps for ordering. This is quietly broken. Two peers on different +machines will have clocks that drift. A peer editing offline and reconnecting +will have a stale wall-clock reading. Under partition, last-write-wins with +wall clocks silently corrupts data. + +P2PCalc uses Hybrid Logical Clocks (HLC), introduced by Kulkarni et al. (2014). +HLC combines physical time (for human-readable timestamps) with a logical +counter (for causal ordering when physical clocks agree). The result: causally +consistent operation ordering across peers without requiring any clock +synchronisation protocol. Two operations from the same peer are always ordered +correctly. Two operations from different peers are ordered by causality when +it exists, and by a deterministic tiebreaker (peer_id) when it does not. + +Every operation carries an HLC timestamp. Every receiving peer updates its +local clock from incoming operations. The invariant is maintained globally +without any coordinator. + +### 2. Multi-Value Register for cell conflicts + +Naive last-write-wins is the wrong model for collaborative spreadsheets. +Consider: Alice and Bob both edit cell A1 simultaneously. Under LWW, one +edit silently disappears with no notification to either user. This is data +loss disguised as a feature. + +P2PCalc implements a Multi-Value Register (MVR) for cell-level conflict +resolution. The MVR tracks all concurrent versions of a cell, those for +which neither version causally precedes the other. When concurrent writes +exist, the cell is marked as a conflict and all competing values are +preserved as candidates. A conflict marker is injected into the spreadsheet +as a comment cell adjacent to the conflicted cell, showing all competing +values. The user can then explicitly resolve which value should win. + +This is the same model Git uses for merge conflicts: preserve all +concurrent edits, surface them to the user, require explicit resolution. +No data is ever silently lost. + +A write that causally follows all existing versions (higher HLC) automatically +resolves any existing conflict by dominating all candidates. This covers +the common case where one peer's edit observes and incorporates the +conflict before writing. + +Three conflict policies are available: + +- `mvr` (default): Multi-Value Register, explicit conflict markers +- `lww`: Last-Write-Wins using HLC, no conflict markers, fast +- `peer`: Deterministic peer-priority (lowest peer_id always wins), useful + for testing and offline-first workflows where one device is authoritative + +### 3. Replicated Growable Array for structural operations + +Cell-value CRDTs are well understood. Row and column insertions and deletions +are harder. Consider: peer A inserts a row at position 3. Peer B simultaneously +deletes row 3. Which row 3 does B mean? The original row 3, or the row that A +just created? + +Operational Transformation (OT) systems handle this with transformation +functions that rewrite operations against each other. OT is notoriously +difficult to implement correctly, and the transformation functions for +spreadsheet structural operations have known correctness issues. + +P2PCalc uses a Replicated Growable Array (RGA) for row and column structure. +RGA tracks the intent of each structural operation rather than its absolute +effect. Concurrent insertions at the same position are deterministically +ordered using peer_id as a tiebreaker, so all peers converge to the same +structural order without coordination. Deletions use tombstones so that +concurrent edits targeting a deleted row can still be applied without +crashing, the deleted row is marked invisible but its position in the +sequence is preserved for conflict resolution. + +### 4. Causal buffering for out-of-order delivery + +GossipSub provides best-effort delivery in an eventually-consistent network. +Operations may arrive out of order, particularly after a peer reconnects or +joins late. Applying operations out of causal order corrupts the CRDT state. + +P2PCalc attaches causal dependency tracking to each operation: every +operation records the op_id of the last operation the issuing peer had +observed on that sheet. The adapter maintains a causal buffer. When an +operation arrives whose declared dependencies have not yet been seen, it is +held in the buffer rather than applied immediately. When the missing +dependency arrives, the buffer is flushed in dependency order. + +This gives P2PCalc causal consistency guarantees on top of GossipSub's +eventual delivery, without requiring a vector clock or a full dependency +graph. + +### 5. Two-phase late-joiner state sync + +When a new peer joins a sheet, it needs the current state. Replaying the +full operation log from the beginning is impractical for long-lived +sessions. + +P2PCalc uses a two-phase approach. On joining, the peer broadcasts a +`SNAPSHOT_REQUEST` operation via GossipSub. Any peer that has state +responds with `SNAPSHOT_CHUNK` operations containing the current cell +state serialised as JSON, chunked into 64KB pieces for libp2p stream +compatibility. The joining peer assembles the chunks and applies the +snapshot. + +Multiple peers may respond simultaneously. P2PCalc selects the response +with the highest operation count (most history) as authoritative. If +counts are equal, the lower peer_id is preferred, ensuring a deterministic +choice without coordination. + +After the snapshot is applied, the joining peer replays any operations +that arrived after the snapshot was taken, using the `op_log_from` field +embedded in the snapshot to know where to start. + +### 6. Local op-log WAL for crash recovery + +P2PCalc persists each operation to a local write-ahead log (WAL) as it is +applied. The WAL uses a 4-byte length-prefixed MessagePack format, the same +encoding used on the wire. + +On restart after a crash, a peer replays its local WAL to recover its last +known state before requesting a network snapshot. This has two benefits: +recovery does not require other peers to be online, and the snapshot request +only needs to cover the delta since the local log's last entry rather than +the full session history. + +This mirrors how distributed databases (CockroachDB, etcd) use WAL-based +recovery to survive crashes without full resync. + +### 7. MessagePack over JSON + +Every operation is serialised with MessagePack rather than JSON. For +spreadsheet collaboration, operations are small and frequent, a busy +editing session may produce tens of operations per second per peer. + +MessagePack is approximately 30-40% smaller than equivalent JSON and 2-3x +faster to encode and decode in Python benchmarks. Over GossipSub, smaller +messages reduce mesh bandwidth consumption and allow the heartbeat interval +to be shortened (currently 1 second rather than the GossipSub default of 2 +seconds) without saturating the network, which directly reduces the latency +between an edit and its propagation to other peers. + +### 8. Operation integrity verification + +Every operation includes a SHA-256 content hash of its key fields (sheet_id, +raw_command, peer_id, op_id). The receiving adapter verifies this hash before +applying the operation. Operations that fail verification are silently +discarded, providing a lightweight integrity guarantee against accidental +corruption or malformed messages from buggy peers. + +The GossipSub message validator hook is used to reject operations that fail +integrity checks at the pub-sub layer, before they are delivered to the +application. This prevents malformed operations from being forwarded to +other peers in the mesh. + +______________________________________________________________________ + +## Architecture + +``` +┌─────────────────────────────────────────────────────────┐ +│ EtherCalc (unchanged) │ +│ Browser <-> WebSocket <-> Node.js worker │ +└───────────────────────┬─────────────────────────────────┘ + │ Redis pub-sub (sc:) + │ +┌───────────────────────┴─────────────────────────────────┐ +│ P2PCalc Adapter (adapter.py) │ +│ │ +│ Redis channel subscriber Redis channel publisher │ +│ (reads local edits) (injects remote edits) │ +│ │ ▲ │ +│ ▼ │ │ +│ OperationFactory Operation.from_bytes() │ +│ (parse SocialCalc command) (deserialise) │ +│ │ │ │ +│ ▼ │ │ +│ SheetCRDT.apply() CausalBuffer │ +│ (local optimistic apply) (hold until deps arrive) │ +│ │ │ │ +└──────────┼───────────────────────────┼──────────────────┘ + │ GossipSub publish │ GossipSub subscribe + ▼ │ +┌──────────────────────────────────────┴──────────────────┐ +│ py-libp2p GossipSub v2.0 (p2p_node.py) │ +│ │ +│ Topic: p2pcalc/ │ +│ Peer discovery: mDNS (LAN) / DHT (WAN) │ +│ Transport: TCP / WebSocket / QUIC │ +│ Security: Noise protocol │ +│ Mux: Yamux │ +└─────────────────────────────────────────────────────────┘ + │ mesh propagation to all peers + ▼ + [Remote P2PCalc peers — same stack] +``` + +______________________________________________________________________ + +## File structure + +``` +examples/p2pcalc/ +├── README.md this file +├── requirements.txt +├── p2pcalc/ +│ ├── __init__.py +│ ├── operation.py operation encoding, HLC, MessagePack protocol +│ ├── crdt.py MVR + RGA CRDT implementations +│ ├── adapter.py EtherCalc Redis <-> GossipSub bridge +│ ├── p2p_node.py libp2p host, GossipSub v2.0, state sync +│ ├── state_sync.py snapshot assembly, op-log WAL persistence +│ └── main.py CLI entrypoint +└── tests/ + ├── __init__.py + └── test_p2pcalc.py 42 unit tests (no network or Redis required) +``` + +______________________________________________________________________ + +## Quick start + +### Prerequisites + +- Python 3.10+ +- Redis running locally: `redis-server` +- EtherCalc running locally: `npm install -g ethercalc && ethercalc` + +### Install + +```bash +cd examples/p2pcalc +pip install -r requirements.txt +``` + +### Run two peers on the same machine + +Terminal 1: + +```bash +python -m p2pcalc.main --room my-sheet --port 4001 +# Output includes: /ip4/127.0.0.1/tcp/4001/p2p/ +``` + +Terminal 2 (paste the multiaddr from Terminal 1): + +```bash +python -m p2pcalc.main --room my-sheet --port 4002 \ + --peer /ip4/127.0.0.1/tcp/4001/p2p/ +``` + +Open `http://localhost:8000/my-sheet` in two browser windows. Edits in +either window propagate to the other via GossipSub with no central server. + +### Run across machines on a LAN + +mDNS peer discovery is active by default. Both machines must be on the +same subnet. Start P2PCalc on each machine with the same `--room` name +and no `--peer` flag. Peers will discover each other automatically. + +### Multiple rooms on one node + +```bash +python -m p2pcalc.main --room room-a --room room-b --port 4001 +``` + +Each room gets its own GossipSub topic and Redis channel subscription. + +### Conflict resolution mode + +```bash +# Multi-Value Register (default), surfaces concurrent conflicts as cell markers +python -m p2pcalc.main --room my-sheet --conflict-policy mvr + +# Last-Write-Wins, faster, silent data loss on concurrent edits +python -m p2pcalc.main --room my-sheet --conflict-policy lww +``` + +### Debug logging + +```bash +python -m p2pcalc.main --room my-sheet --debug +``` + +______________________________________________________________________ + +## Run tests + +No network, Redis, or EtherCalc required. + +```bash +cd examples/p2pcalc +python -m pytest tests/test_p2pcalc.py -v +# Expected: 42 passed +``` + +Test coverage includes HLC monotonicity and causal ordering, operation +parsing for all SocialCalc command types, MessagePack serialisation +roundtrip, integrity verification, MVR conflict detection and resolution, +RGA concurrent structural operation merging, SheetCRDT deduplication and +op-log, snapshot assembly and race resolution, and multi-peer convergence +simulation. + +______________________________________________________________________ + +## What is not yet implemented + +- DHT-based WAN peer discovery (mDNS only covers LAN today) +- IPFS snapshot storage for durable shared state +- Benchmarking harness comparing latency against centralised EtherCalc +- WebRTC transport for browser-to-browser operation (no Python needed) + +These are planned for subsequent PRs in the DMP 2026 timeline. + +______________________________________________________________________ + +## References + +- Kulkarni et al., "Logical Physical Clocks and Consistent Snapshots in + Globally Distributed Databases", 2014 (HLC) +- Shapiro et al., "Conflict-Free Replicated Data Types", INRIA 2011 (CRDT) +- Roh et al., "Replicated Abstract Data Types: Building Blocks for + Collaborative Applications", 2011 (RGA) +- Tan et al., "From SocialCalc to EtherCalc", AOSA Volume 2, 2012 + (EtherCalc internals and Redis pub-sub architecture) +- libp2p GossipSub v1.1 spec: https://github.com/libp2p/specs/tree/master/pubsub/gossipsub diff --git a/examples/p2pcalc/conftest.py b/examples/p2pcalc/conftest.py new file mode 100644 index 000000000..4ea0c6cdd --- /dev/null +++ b/examples/p2pcalc/conftest.py @@ -0,0 +1,4 @@ +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).parent)) diff --git a/examples/p2pcalc/p2pcalc/__init__.py b/examples/p2pcalc/p2pcalc/__init__.py new file mode 100644 index 000000000..0fbf8c0d0 --- /dev/null +++ b/examples/p2pcalc/p2pcalc/__init__.py @@ -0,0 +1,10 @@ +""" +P2PCalc - Decentralised EtherCalc collaboration over py-libp2p GossipSub. + +Core components: +- operation: protocol, HLC, operation encoding +- crdt: MVR and RGA CRDT implementations +- p2p_node: libp2p + GossipSub networking +- adapter: EtherCalc Redis bridge +- state_sync: snapshot and WAL recovery +""" diff --git a/examples/p2pcalc/p2pcalc/adapter.py b/examples/p2pcalc/p2pcalc/adapter.py new file mode 100644 index 000000000..a97e2157e --- /dev/null +++ b/examples/p2pcalc/p2pcalc/adapter.py @@ -0,0 +1,368 @@ +# p2pcalc/adapter.py — EtherCalc/Redis to libp2p GossipSub adapter +# Author: Dashpreet Singh +# +# This is the core innovation of the project. +# +# EtherCalc uses Redis pub-sub to broadcast SocialCalc commands between +# its Node.js workers. Each room publishes to channel "sc:". +# The command format is a raw SocialCalc command string: +# "set A1 value 42" +# "set B2 formula =SUM(A1:A5)" +# +# This adapter: +# 1. Subscribes to EtherCalc's Redis channel for a room +# 2. Translates each command into a P2PCalc Operation +# 3. Publishes the Operation to GossipSub (replaces central server) +# 4. Receives Operations from GossipSub peers +# 5. Injects them back into EtherCalc's Redis channel +# 6. EtherCalc's existing Socket.io layer delivers to browsers +# +# The UI (EtherCalc/SocialCalc in the browser) is NEVER touched. +# The server-side Node.js EtherCalc worker is NEVER touched. +# We intercept only at the Redis pub-sub layer. +# +# ADDITIONAL INNOVATIONS: +# - Operation deduplication at the Redis inject layer +# (prevents echo loops when our own ops come back via GossipSub) +# - Causal buffering: ops with unmet dependencies are buffered +# until their parents arrive (prevents out-of-order application) +# - Presence injection: peer cursor positions are injected as +# EtherCalc "cursor" commands so users see each other's positions +# - Conflict markers: when MVR detects a conflict, a comment cell +# is injected into EtherCalc showing the competing values + +from __future__ import annotations + +import asyncio +from collections import defaultdict +import logging + +import redis.asyncio as aioredis + +from .crdt import ConflictPolicy +from .operation import Operation +from .p2p_node import P2PCalcNode + +logger = logging.getLogger("p2pcalc.adapter") + +# EtherCalc Redis channel pattern: "sc:" +ETHERCALC_CHANNEL_PREFIX = "sc" + +# Our own operations that we inject back into Redis get tagged with this +# prefix in a side-channel to detect echo loops +ECHO_TAG_PREFIX = "p2pcalc_injected:" + + +class EtherCalcAdapter: + """ + Bridges EtherCalc's Redis pub-sub with the P2PCalc GossipSub network. + + One adapter instance handles one EtherCalc room (sheet). + Multiple adapters can run in the same process for multiple rooms. + """ + + def __init__( + self, + room_name: str, + node: P2PCalcNode, + redis_url: str = "redis://localhost:6379", + conflict_policy: ConflictPolicy = ConflictPolicy.MULTI_VALUE, + ): + self._room = room_name + self._node = node + self._redis_url = redis_url + self._sheet_id = room_name + + # Redis clients — separate for pub and sub + self._redis_pub: aioredis.Redis | None = None + self._redis_sub: aioredis.client.PubSub | None = None + + # Dedup: track op_ids we injected so we don't re-broadcast them + self._injected_op_ids: set[str] = set() + self._injected_commands: set[str] = set() # raw command strings + + # Causal buffer: ops waiting for their depends_on to arrive + self._causal_buffer: dict[str, list[Operation]] = defaultdict(list) + + # Metrics + self._stats = { + "redis_received": 0, + "p2p_published": 0, + "p2p_received": 0, + "redis_injected": 0, + "duplicates_dropped": 0, + "causal_buffered": 0, + "conflicts_detected": 0, + } + + self._running = False + + # ------------------------------------------------------------------ + # Startup + # ------------------------------------------------------------------ + + async def start(self) -> None: + self._redis_pub = await aioredis.from_url( + self._redis_url, decode_responses=True + ) + redis_client = await aioredis.from_url(self._redis_url, decode_responses=True) + self._redis_sub = redis_client.pubsub() + + # Subscribe to EtherCalc's Redis channel for this room + channel = f"{ETHERCALC_CHANNEL_PREFIX}:{self._room}" + await self._redis_sub.subscribe(channel) + logger.info("Adapter started for room '%s' (channel=%s)", self._room, channel) + + # Join the P2P network for this sheet + await self._node.join_sheet( + sheet_id=self._sheet_id, + on_operation=self._on_p2p_operation, + request_snapshot=True, + ) + + self._running = True + + # Run both pumps concurrently + await asyncio.gather( + self._redis_pump(), + self._heartbeat_pump(), + ) + + async def stop(self) -> None: + self._running = False + if self._redis_sub: + await self._redis_sub.unsubscribe() + if self._redis_pub: + await self._redis_pub.aclose() + logger.info("Adapter stopped for room '%s'", self._room) + + # ------------------------------------------------------------------ + # Redis -> P2P path (outbound) + # ------------------------------------------------------------------ + + async def _redis_pump(self) -> None: + """Read commands from EtherCalc Redis, publish to GossipSub.""" + async for message in self._redis_sub.listen(): # type: ignore[union-attr] + if not self._running: + break + if message["type"] != "message": + continue + + raw = message["data"] + + # Skip commands we injected ourselves — echo loop prevention + if raw.startswith(ECHO_TAG_PREFIX): + continue + if raw in self._injected_commands: + self._injected_commands.discard(raw) + self._stats["duplicates_dropped"] += 1 + continue + + self._stats["redis_received"] += 1 + logger.debug("Redis -> P2P: %r", raw[:80]) + + await self._node.publish_command(self._sheet_id, raw) + self._stats["p2p_published"] += 1 + + # Check for conflicts after apply + conflicts = self._node.get_conflicts(self._sheet_id) + if conflicts: + self._stats["conflicts_detected"] += len(conflicts) + for conflict_cell in conflicts: + await self._inject_conflict_marker(conflict_cell) + + # ------------------------------------------------------------------ + # P2P -> Redis path (inbound) + # ------------------------------------------------------------------ + + async def _on_p2p_operation(self, op: Operation) -> None: + """Receive an Operation from GossipSub, inject into EtherCalc Redis.""" + self._stats["p2p_received"] += 1 + + # Check causal dependencies + if op.depends_on: + sheet = self._node.get_sheet_state(self._sheet_id) + if sheet: + seen_ids = set() + for past_op in sheet.op_log_since(0): + seen_ids.add(past_op.op_id) + unmet = [dep for dep in op.depends_on if dep not in seen_ids] + if unmet: + logger.debug("Op %s buffered — waiting for %s", op.op_id[:8], unmet) + for dep in unmet: + self._causal_buffer[dep].append(op) + self._stats["causal_buffered"] += 1 + return + + await self._inject_to_redis(op) + + # Flush any ops that were waiting for this op + await self._flush_causal_buffer(op.op_id) + + async def _inject_to_redis(self, op: Operation) -> None: + """Inject a P2P operation's raw command into EtherCalc Redis.""" + if not op.raw_command or not self._redis_pub: + return + + if op.op_id in self._injected_op_ids: + self._stats["duplicates_dropped"] += 1 + return + + self._injected_op_ids.add(op.op_id) + # Bound the set size + if len(self._injected_op_ids) > 10000: + self._injected_op_ids = set(list(self._injected_op_ids)[-5000:]) + + # Track so redis_pump doesn't re-broadcast it + self._injected_commands.add(op.raw_command) + + channel = f"{ETHERCALC_CHANNEL_PREFIX}:{self._room}" + await self._redis_pub.publish(channel, op.raw_command) + self._stats["redis_injected"] += 1 + + logger.debug( + "P2P -> Redis: peer=%s cmd=%r", + op.peer_id[:8], + op.raw_command[:60], + ) + + async def _flush_causal_buffer(self, arrived_op_id: str) -> None: + """Flush operations that were waiting for arrived_op_id.""" + waiting = self._causal_buffer.pop(arrived_op_id, []) + for op in waiting: + logger.debug( + "Flushing buffered op %s (dependency %s arrived)", + op.op_id[:8], + arrived_op_id[:8], + ) + await self._inject_to_redis(op) + await self._flush_causal_buffer(op.op_id) + + # ------------------------------------------------------------------ + # Conflict markers + # ------------------------------------------------------------------ + + async def _inject_conflict_marker(self, conflict_cell) -> None: + """ + Inject a conflict marker into EtherCalc as a comment cell. + + When MVR detects concurrent conflicting edits, we inject a + special comment into the cell adjacent to the conflict so users + can see and resolve it directly in the spreadsheet UI. + """ + if not self._redis_pub: + return + + coord = conflict_cell.coord + values = [v.value for v in conflict_cell.candidates[:3]] + marker = f"CONFLICT: {' | '.join(values)}" + + # Find the column letter to place the marker one column to the right + col_letter = "".join(c for c in coord if c.isalpha()) + row_num = "".join(c for c in coord if c.isdigit()) + next_col = chr(ord(col_letter[-1]) + 1) if col_letter else "B" + marker_coord = f"{next_col}{row_num}" + + # Inject as a cell comment (SocialCalc format) + command = f"set {marker_coord} value {marker}" + channel = f"{ETHERCALC_CHANNEL_PREFIX}:{self._room}" + await self._redis_pub.publish(channel, command) + logger.info("Injected conflict marker at %s for %s", marker_coord, coord) + + # ------------------------------------------------------------------ + # Heartbeat + # ------------------------------------------------------------------ + + async def _heartbeat_pump(self) -> None: + """Send periodic heartbeats to advertise presence.""" + while self._running: + await asyncio.sleep(10) + try: + await self._node.send_heartbeat(self._sheet_id) + except Exception as e: + logger.debug("Heartbeat error: %s", e) + + # ------------------------------------------------------------------ + # Stats + # ------------------------------------------------------------------ + + def stats(self) -> dict: + return { + **self._stats, + "room": self._room, + "peer_id": self._node.peer_id[:12], + "p2p_addrs": self._node.get_listen_addrs(), + "presence": self._node.get_presence(self._sheet_id), + "conflicts_pending": len(self._node.get_conflicts(self._sheet_id)), + } + + +# --------------------------------------------------------------------------- +# Multi-room adapter manager +# --------------------------------------------------------------------------- + + +class AdapterManager: + """ + Manages multiple EtherCalc room adapters on a single P2P node. + + One P2P node, many rooms — each room gets its own GossipSub topic + and Redis channel subscription. + """ + + def __init__( + self, + redis_url: str = "redis://localhost:6379", + p2p_port: int = 0, + known_peers: list[str] | None = None, + conflict_policy: ConflictPolicy = ConflictPolicy.MULTI_VALUE, + ): + self._redis_url = redis_url + self._conflict_policy = conflict_policy + self._node = P2PCalcNode( + port=p2p_port, + known_peers=known_peers or [], + conflict_policy=conflict_policy, + ) + self._adapters: dict[str, EtherCalcAdapter] = {} + self._started = False + + async def start(self) -> str: + peer_id = await self._node.start() + self._started = True + logger.info("AdapterManager started: peer_id=%s", peer_id) + logger.info("Listening on: %s", self._node.get_listen_addrs()) + return peer_id + + async def add_room(self, room_name: str) -> EtherCalcAdapter: + if room_name in self._adapters: + return self._adapters[room_name] + adapter = EtherCalcAdapter( + room_name=room_name, + node=self._node, + redis_url=self._redis_url, + conflict_policy=self._conflict_policy, + ) + self._adapters[room_name] = adapter + asyncio.create_task(adapter.start()) + logger.info("Added room: %s", room_name) + return adapter + + async def remove_room(self, room_name: str) -> None: + adapter = self._adapters.pop(room_name, None) + if adapter: + await adapter.stop() + + def all_stats(self) -> dict: + return { + "peer_id": self._node.peer_id, + "listen_addrs": self._node.get_listen_addrs(), + "rooms": { + name: adapter.stats() for name, adapter in self._adapters.items() + }, + } + + async def stop(self) -> None: + for adapter in self._adapters.values(): + await adapter.stop() + await self._node.stop() diff --git a/examples/p2pcalc/p2pcalc/crdt.py b/examples/p2pcalc/p2pcalc/crdt.py new file mode 100644 index 000000000..8e1343dde --- /dev/null +++ b/examples/p2pcalc/p2pcalc/crdt.py @@ -0,0 +1,397 @@ +# p2pcalc/crdt.py — CRDT conflict resolution for P2PCalc +# Author: Dashpreet Singh +# +# INNOVATION: We implement a two-phase CRDT approach: +# +# Phase 1 (cell values): Multi-Value Register (MVR) +# Unlike naive Last-Write-Wins (LWW) which silently loses data, +# MVR tracks ALL concurrent values for a cell. When concurrent +# edits happen, both values are preserved and surfaced to the user +# as a conflict marker (like Git merge conflicts, but for cells). +# The user explicitly resolves — no silent data loss. +# +# Phase 2 (structural ops): Replicated Growable Array (RGA) for rows/cols +# Row/column insertions and deletions are tracked as a CRDT sequence. +# Concurrent insertions use peer_id as a tiebreaker for determinism. +# Deletions use tombstones so concurrent edits to a deleted row can +# still be applied without crashing. +# +# This is significantly more principled than the "last-write-wins using +# timestamps" approach most collaborative editors use, which breaks under +# network partition. + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +import logging + +from .operation import HLC, Operation, OpType + +logger = logging.getLogger("p2pcalc.crdt") + +_clock = HLC() + + +# --------------------------------------------------------------------------- +# Cell state +# --------------------------------------------------------------------------- + + +class ConflictPolicy(Enum): + LAST_WRITE_WINS = "lww" # Phase 1 baseline — fast, simple + MULTI_VALUE = "mvr" # Phase 2 — safe, surfaced to user + PEER_PRIORITY = "peer" # Deterministic: lowest peer_id wins + + +@dataclass +class CellVersion: + """A single versioned value for a cell.""" + + value: str + formula: str + formats: dict # attr -> value + hlc: tuple[int, int] + peer_id: str + op_id: str + + +@dataclass +class CellState: + """ + CRDT state for a single cell. + + In Phase 1 (LWW): only the winning version is kept. + In Phase 2 (MVR): all concurrent versions are kept as candidates. + A cell is in "conflict" when len(candidates) > 1. + """ + + coord: str # e.g. "A1" + candidates: list[CellVersion] = field(default_factory=list) + resolved_value: str | None = None # set by user during conflict resolution + is_conflict: bool = False + + @property + def current_value(self) -> str: + if self.is_conflict and self.resolved_value is not None: + return self.resolved_value + if self.candidates: + return self.candidates[0].value + return "" + + @property + def current_formula(self) -> str: + if self.candidates: + return self.candidates[0].formula + return "" + + +# --------------------------------------------------------------------------- +# Multi-Value Register (cell-level CRDT) +# --------------------------------------------------------------------------- + + +class MultiValueRegister: + """ + Cell-level CRDT using Multi-Value Register semantics. + + A write dominates another if it causally follows it (HLC comparison). + Concurrent writes (neither dominates the other) produce a conflict. + + This is mathematically equivalent to a state-based CRDT G-Set of + (value, version) pairs, merged by taking the union and pruning + dominated versions. + """ + + def __init__(self, policy: ConflictPolicy = ConflictPolicy.MULTI_VALUE): + self._policy = policy + self._cells: dict[str, CellState] = {} + + def apply(self, op: Operation) -> CellState: + """Apply a cell operation and return the resulting cell state.""" + coord = op.payload.get("cell", "") + if not coord: + return CellState(coord="") + + state = self._cells.get(coord, CellState(coord=coord)) + + new_version = CellVersion( + value=op.payload.get("value", ""), + formula=op.payload.get("formula", ""), + formats={op.payload.get("attr", ""): op.payload.get("value", "")} + if op.op_type == OpType.SET_CELL_FORMAT + else {}, + hlc=op.hlc, + peer_id=op.peer_id, + op_id=op.op_id, + ) + + if op.op_type == OpType.CLEAR_CELL: + state.candidates = [] + state.is_conflict = False + state.resolved_value = None + self._cells[coord] = state + return state + + if self._policy == ConflictPolicy.LAST_WRITE_WINS: + state = self._apply_lww(state, new_version) + elif self._policy == ConflictPolicy.MULTI_VALUE: + state = self._apply_mvr(state, new_version) + elif self._policy == ConflictPolicy.PEER_PRIORITY: + state = self._apply_peer_priority(state, new_version) + + self._cells[coord] = state + return state + + def _apply_lww(self, state: CellState, new_ver: CellVersion) -> CellState: + """Last-Write-Wins: higher HLC always wins.""" + if not state.candidates: + state.candidates = [new_ver] + else: + existing = state.candidates[0] + cmp = _clock.compare(new_ver.hlc, existing.hlc) + if cmp > 0: + state.candidates = [new_ver] + elif cmp == 0: + # Tie: use peer_id as deterministic tiebreaker + if new_ver.peer_id < existing.peer_id: + state.candidates = [new_ver] + state.is_conflict = False + return state + + def _apply_mvr(self, state: CellState, new_ver: CellVersion) -> CellState: + """Multi-Value Register: preserve all concurrent versions.""" + survivors = [] + new_dominates_all = True + + for existing in state.candidates: + cmp = _clock.compare(new_ver.hlc, existing.hlc) + if cmp > 0: + # new_ver causally dominates existing — prune existing + pass + elif cmp < 0: + # existing causally dominates new_ver — new_ver is stale + new_dominates_all = False + survivors.append(existing) + else: + # Concurrent — keep both + survivors.append(existing) + + if new_dominates_all or not state.candidates: + survivors.append(new_ver) + + # Sort by peer_id for deterministic ordering + survivors.sort(key=lambda v: (v.hlc[0], v.hlc[1], v.peer_id)) + state.candidates = survivors + state.is_conflict = len(survivors) > 1 + if state.is_conflict: + logger.info( + "Conflict at %s: %d concurrent values from peers %s", + state.coord, + len(survivors), + [v.peer_id[:8] for v in survivors], + ) + return state + + def _apply_peer_priority(self, state: CellState, new_ver: CellVersion) -> CellState: + """Deterministic: lexicographically smallest peer_id always wins.""" + if not state.candidates: + state.candidates = [new_ver] + else: + existing = state.candidates[0] + if new_ver.peer_id <= existing.peer_id: + state.candidates = [new_ver] + state.is_conflict = False + return state + + def resolve_conflict(self, coord: str, chosen_op_id: str) -> CellState: + """User explicitly picks one candidate as the resolved value.""" + state = self._cells.get(coord) + if not state: + return CellState(coord=coord) + chosen = next((v for v in state.candidates if v.op_id == chosen_op_id), None) + if chosen: + state.resolved_value = chosen.value + state.is_conflict = False + state.candidates = [chosen] + return state + + def get_conflicts(self) -> list[CellState]: + return [s for s in self._cells.values() if s.is_conflict] + + def get_cell(self, coord: str) -> CellState | None: + return self._cells.get(coord) + + def snapshot(self) -> dict[str, dict]: + """Export current state as a serialisable dict.""" + result = {} + for coord, state in self._cells.items(): + if state.candidates: + best = state.candidates[0] + result[coord] = { + "value": best.value, + "formula": best.formula, + "formats": best.formats, + "is_conflict": state.is_conflict, + "candidates": len(state.candidates), + } + return result + + +# --------------------------------------------------------------------------- +# RGA for structural operations (row/column insertions + deletions) +# --------------------------------------------------------------------------- + + +@dataclass +class RGANode: + """A node in the Replicated Growable Array.""" + + index: int # Original row or column index at insertion time + peer_id: str # Peer that inserted this + hlc: tuple[int, int] + is_tombstone: bool = False # Soft delete — never remove from array + + +class StructuralCRDT: + """ + RGA-based CRDT for row/column insertions and deletions. + + Key innovation: we track the INTENT of structural operations + (insert row 3) rather than their absolute effect (rows 4..N shift down). + This allows concurrent structural ops to be merged without the ambiguity + of "which row 3 did you mean?" problem that plagues OT-based systems. + + Concurrent insertions at the same position use peer_id as a tiebreaker + to deterministically order them. + """ + + def __init__(self): + self._rows: list[RGANode] = [] + self._cols: list[RGANode] = [] + + def apply(self, op: Operation) -> None: + if op.op_type == OpType.INSERT_ROW: + self._insert(self._rows, op.payload["row"], op.peer_id, op.hlc) + elif op.op_type == OpType.DELETE_ROW: + self._delete(self._rows, op.payload["row"]) + elif op.op_type == OpType.INSERT_COL: + self._insert(self._cols, op.payload.get("col_index", 0), op.peer_id, op.hlc) + elif op.op_type == OpType.DELETE_COL: + self._delete(self._cols, op.payload.get("col_index", 0)) + + def _insert( + self, + seq: list[RGANode], + idx: int, + peer_id: str, + hlc: tuple[int, int], + ) -> None: + node = RGANode(index=idx, peer_id=peer_id, hlc=hlc) + # Find insertion position: after all nodes with same or earlier index, + # with concurrent nodes ordered by peer_id (deterministic tiebreak) + insert_pos = len(seq) + for i, existing in enumerate(seq): + if existing.index > idx: + insert_pos = i + break + if existing.index == idx: + # Concurrent — lower peer_id goes first (arbitrary but deterministic) + if peer_id < existing.peer_id: + insert_pos = i + break + seq.insert(insert_pos, node) + + def _delete(self, seq: list[RGANode], idx: int) -> None: + for node in seq: + if node.index == idx and not node.is_tombstone: + node.is_tombstone = True + return + + def visible_rows(self) -> list[int]: + """Return the current visible row indices in order.""" + return [n.index for n in self._rows if not n.is_tombstone] + + def visible_cols(self) -> list[int]: + return [n.index for n in self._cols if not n.is_tombstone] + + def snapshot(self) -> dict: + return { + "rows": [ + {"idx": n.index, "peer": n.peer_id, "deleted": n.is_tombstone} + for n in self._rows + ], + "cols": [ + {"idx": n.index, "peer": n.peer_id, "deleted": n.is_tombstone} + for n in self._cols + ], + } + + +# --------------------------------------------------------------------------- +# Unified sheet CRDT state +# --------------------------------------------------------------------------- + + +class SheetCRDT: + """ + Complete CRDT state for a single spreadsheet sheet. + + Combines: + - MultiValueRegister for all cell mutations + - StructuralCRDT for row/column structure + - Op-log for late-joiner state reconstruction + """ + + def __init__( + self, + sheet_id: str, + policy: ConflictPolicy = ConflictPolicy.MULTI_VALUE, + ): + self.sheet_id = sheet_id + self.cells = MultiValueRegister(policy=policy) + self.structure = StructuralCRDT() + self._op_log: list[Operation] = [] + self._seen_op_ids: set[str] = set() + + def apply(self, op: Operation) -> bool: + """Apply an operation. Returns False if already seen (duplicate).""" + if op.op_id in self._seen_op_ids: + return False + self._seen_op_ids.add(op.op_id) + self._op_log.append(op) + + if op.op_type in ( + OpType.SET_CELL_VALUE, + OpType.SET_CELL_FORMULA, + OpType.SET_CELL_FORMAT, + OpType.SET_RANGE_FORMAT, + OpType.CLEAR_CELL, + ): + self.cells.apply(op) + elif op.op_type in ( + OpType.INSERT_ROW, + OpType.DELETE_ROW, + OpType.INSERT_COL, + OpType.DELETE_COL, + ): + self.structure.apply(op) + + return True + + def get_conflicts(self) -> list[CellState]: + return self.cells.get_conflicts() + + def op_log_since(self, from_index: int) -> list[Operation]: + return self._op_log[from_index:] + + def op_log_length(self) -> int: + return len(self._op_log) + + def snapshot(self) -> dict: + return { + "sheet_id": self.sheet_id, + "cells": self.cells.snapshot(), + "structure": self.structure.snapshot(), + "op_count": len(self._op_log), + } diff --git a/examples/p2pcalc/p2pcalc/main.py b/examples/p2pcalc/p2pcalc/main.py new file mode 100644 index 000000000..a0d7315d5 --- /dev/null +++ b/examples/p2pcalc/p2pcalc/main.py @@ -0,0 +1,153 @@ +#!/usr/bin/env python3 +# p2pcalc/main.py — P2PCalc entrypoint +# Author: Dashpreet Singh +# +# Usage: +# # Start a node, join room "my-sheet" +# python -m p2pcalc.main --room my-sheet --port 4001 +# +# # Second peer, connecting to first +# python -m p2pcalc.main --room my-sheet --port 4002 \ +# --peer /ip4/127.0.0.1/tcp/4001/p2p/ +# +# # Multiple rooms on one node +# python -m p2pcalc.main --room room1 --room room2 --port 4001 + +import argparse +import asyncio +import json +import logging +import signal +import sys + +from .adapter import AdapterManager +from .crdt import ConflictPolicy + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(name)s] %(levelname)s: %(message)s", +) +logger = logging.getLogger("p2pcalc") + + +async def run(args: argparse.Namespace) -> None: + policy_map = { + "lww": ConflictPolicy.LAST_WRITE_WINS, + "mvr": ConflictPolicy.MULTI_VALUE, + "peer": ConflictPolicy.PEER_PRIORITY, + } + policy = policy_map.get(args.conflict_policy, ConflictPolicy.MULTI_VALUE) + + manager = AdapterManager( + redis_url=args.redis, + p2p_port=args.port, + known_peers=args.peer or [], + conflict_policy=policy, + ) + + peer_id = await manager.start() + print("\nP2PCalc node started") + print(f"Peer ID : {peer_id}") + print(f"Addrs : {manager._node.get_listen_addrs()}") + print(f"Redis : {args.redis}") + print(f"Policy : {policy.value}") + + for room in args.room: + await manager.add_room(room) + print(f"Room : {room} (topic=p2pcalc/{room})") + + print("\nReady. Ctrl-C to stop.\n") + + # Periodic stats logging + async def stats_loop() -> None: + while True: + await asyncio.sleep(30) + stats = manager.all_stats() + logger.info("Stats: %s", json.dumps(stats, indent=2)) + + stop_event = asyncio.Event() + + def _signal_handler(*args: object) -> None: + print("\nShutting down...") + stop_event.set() + + # asyncio.loop.add_signal_handler is not available on Windows; + # fall back to the stdlib signal module which works on all platforms. + if sys.platform != "win32": + loop = asyncio.get_running_loop() + loop.add_signal_handler(signal.SIGINT, _signal_handler) + loop.add_signal_handler(signal.SIGTERM, _signal_handler) + else: + signal.signal(signal.SIGINT, lambda _s, _f: _signal_handler()) + signal.signal(signal.SIGTERM, lambda _s, _f: _signal_handler()) + + stats_task = asyncio.create_task(stats_loop()) + + await stop_event.wait() + + stats_task.cancel() + await manager.stop() + print("Stopped.") + + +def main() -> None: + parser = argparse.ArgumentParser( + description="P2PCalc — Decentralised EtherCalc via py-libp2p GossipSub" + ) + parser.add_argument( + "--room", + "-r", + action="append", + default=[], + metavar="ROOM_NAME", + help="EtherCalc room name(s) to bridge (can repeat)", + ) + parser.add_argument( + "--port", + "-p", + type=int, + default=0, + help="TCP port for libp2p (0 = random)", + ) + parser.add_argument( + "--peer", + action="append", + default=[], + metavar="MULTIADDR", + help="Known peer multiaddr(s) to connect to on startup", + ) + parser.add_argument( + "--redis", + default="redis://localhost:6379", + help="Redis URL (default: redis://localhost:6379)", + ) + parser.add_argument( + "--conflict-policy", + choices=["lww", "mvr", "peer"], + default="mvr", + help=( + "Conflict resolution: lww=last-write-wins, " + "mvr=multi-value (default), peer=peer-priority" + ), + ) + parser.add_argument( + "--debug", + action="store_true", + help="Enable debug logging", + ) + + args = parser.parse_args() + + if args.debug: + logging.getLogger().setLevel(logging.DEBUG) + + if not args.room: + print("Error: at least one --room is required") + parser.print_help() + sys.exit(1) + + asyncio.run(run(args)) + + +if __name__ == "__main__": + main() diff --git a/examples/p2pcalc/p2pcalc/operation.py b/examples/p2pcalc/p2pcalc/operation.py new file mode 100644 index 000000000..89db0b2d1 --- /dev/null +++ b/examples/p2pcalc/p2pcalc/operation.py @@ -0,0 +1,405 @@ +# p2pcalc/operation.py — Spreadsheet operation encoding protocol +# Author: Dashpreet Singh +# IIT Jammu, B.Tech CSE 2024-2028 +# +# Design principles: +# - Operations are the unit of truth, never full state +# - Every operation is deterministically replayable +# - MessagePack for compact binary serialisation (faster than JSON on wire) +# - Hybrid Logical Clocks (HLC) for causally-consistent ordering across peers +# without requiring clock synchronisation +# +# SocialCalc command format (what EtherCalc actually sends): +# "set A1 value 42" +# "set B2 formula =SUM(A1:A5)" +# "set A1:B5 bgcolor #ff0000" +# "insertrow 3 1" +# "deletecol C 2" +# +# We wrap these commands in a typed envelope and add HLC timestamps, +# peer identity, causal dependencies, and a content hash for integrity. + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import IntEnum +import hashlib +import time +from typing import Any, cast +import uuid + +import msgpack + +# --------------------------------------------------------------------------- +# Hybrid Logical Clock +# --------------------------------------------------------------------------- + + +class HLC: + """ + Hybrid Logical Clock for causal ordering without clock sync. + + HLC combines physical time (for human readability) with a logical + counter (for causal ordering when physical clocks agree). This gives + us causally-consistent ordering across peers even with clock drift, + which pure Lamport clocks or pure wall clocks cannot guarantee alone. + + Reference: Kulkarni et al., "Logical Physical Clocks" (2014) + """ + + def __init__(self): + self._physical: int = 0 # milliseconds + self._logical: int = 0 + + def now(self) -> tuple[int, int]: + """Generate a new HLC timestamp. Returns (physical_ms, logical).""" + pt = int(time.time() * 1000) + if pt > self._physical: + self._physical = pt + self._logical = 0 + else: + self._logical += 1 + return (self._physical, self._logical) + + def update(self, remote_physical: int, remote_logical: int) -> tuple[int, int]: + """Receive a remote HLC timestamp and advance local clock.""" + pt = int(time.time() * 1000) + self._physical = max(self._physical, remote_physical, pt) + if self._physical == remote_physical == pt: + self._logical = max(self._logical, remote_logical) + 1 + elif self._physical == remote_physical: + self._logical = max(self._logical, remote_logical) + 1 + elif self._physical == pt: + self._logical += 1 + else: + self._logical = 0 + return (self._physical, self._logical) + + def compare( + self, + a: tuple[int, int], + b: tuple[int, int], + ) -> int: + """Compare two HLC timestamps. Returns -1, 0, or 1.""" + if a[0] != b[0]: + return -1 if a[0] < b[0] else 1 + if a[1] != b[1]: + return -1 if a[1] < b[1] else 1 + return 0 + + +# Module-level singleton clock +_clock = HLC() + + +# --------------------------------------------------------------------------- +# Operation types +# --------------------------------------------------------------------------- + + +class OpType(IntEnum): + # Cell operations (mapped from SocialCalc command strings) + SET_CELL_VALUE = 0x01 + SET_CELL_FORMULA = 0x02 + SET_CELL_FORMAT = 0x03 + CLEAR_CELL = 0x04 + + # Range operations + SET_RANGE_FORMAT = 0x10 + COPY_RANGE = 0x11 + MOVE_RANGE = 0x12 + + # Structural operations + INSERT_ROW = 0x20 + DELETE_ROW = 0x21 + INSERT_COL = 0x22 + DELETE_COL = 0x23 + + # Sheet operations + SORT_SHEET = 0x30 + RENAME_SHEET = 0x31 + + # Control + SNAPSHOT_REQUEST = 0xF0 # late joiner asks for current state + SNAPSHOT_CHUNK = 0xF1 # response: one chunk of snapshot + HEARTBEAT = 0xFF # peer presence signal + + +# --------------------------------------------------------------------------- +# Core Operation dataclass +# --------------------------------------------------------------------------- + + +@dataclass +class Operation: + """ + A single spreadsheet operation in the P2PCalc protocol. + + Every mutation flowing through the GossipSub topic is an Operation. + Operations are: + - Totally ordered within a sheet via HLC timestamps + - Deduplicated by op_id across the network + - Self-describing: contain enough info to apply or undo + - Compact: serialised with MessagePack (not JSON) + """ + + # Identity + op_id: str # UUID4 — globally unique, used for dedup + peer_id: str # libp2p peer ID of originating peer + sheet_id: str # EtherCalc room name (e.g. "my-spreadsheet") + + # Causality + hlc_physical: int # HLC physical component (ms) + hlc_logical: int # HLC logical component + + # Content + op_type: int # OpType enum value + raw_command: str # Original SocialCalc command string (audit trail) + payload: dict # Structured payload (type-specific fields) + + # Integrity + content_hash: str = "" # SHA-256 of (sheet_id + raw_command + peer_id) + + # Causal dependency (optional, for CRDT causal tracking) + depends_on: list[str] = field(default_factory=list) + + def __post_init__(self): + if not self.content_hash: + self.content_hash = self._compute_hash() + + def _compute_hash(self) -> str: + raw = f"{self.sheet_id}:{self.raw_command}:{self.peer_id}:{self.op_id}" + return hashlib.sha256(raw.encode()).hexdigest()[:16] + + @property + def hlc(self) -> tuple[int, int]: + return (self.hlc_physical, self.hlc_logical) + + def to_bytes(self) -> bytes: + """Serialise to MessagePack bytes for wire transmission.""" + d = { + "i": self.op_id, + "p": self.peer_id, + "s": self.sheet_id, + "tp": self.hlc_physical, + "tl": self.hlc_logical, + "t": self.op_type, + "c": self.raw_command, + "d": self.payload, + "h": self.content_hash, + "x": self.depends_on, + } + result = msgpack.packb(d, use_bin_type=True) + assert result is not None + return result + + @classmethod + def from_bytes(cls, data: bytes) -> Operation: + """Deserialise from MessagePack bytes.""" + d = cast(dict[str, Any], msgpack.unpackb(data, raw=False)) + return cls( + op_id=d["i"], + peer_id=d["p"], + sheet_id=d["s"], + hlc_physical=d["tp"], + hlc_logical=d["tl"], + op_type=d["t"], + raw_command=d["c"], + payload=d["d"], + content_hash=d["h"], + depends_on=d.get("x", []), + ) + + def verify_integrity(self) -> bool: + return self.content_hash == self._compute_hash() + + +# --------------------------------------------------------------------------- +# Operation factory — parse SocialCalc command strings into Operations +# --------------------------------------------------------------------------- + + +class OperationFactory: + """ + Parse EtherCalc/SocialCalc command strings into Operation objects. + + SocialCalc sends human-readable command strings over its channel. + We parse these, classify them, extract structured payload, and wrap + in a P2PCalc Operation with HLC timestamp and peer identity. + + This factory is the critical translation layer between EtherCalc's + existing protocol and our p2p layer. No UI changes needed. + """ + + def __init__(self, peer_id: str): + self._peer_id = peer_id + + def from_socialcalc_command( + self, + command: str, + sheet_id: str, + depends_on: list[str] | None = None, + ) -> Operation: + """Parse a raw SocialCalc command string into a typed Operation.""" + command = command.strip() + hlc_p, hlc_l = _clock.now() + + op_type, payload = self._classify(command) + + return Operation( + op_id=str(uuid.uuid4()), + peer_id=self._peer_id, + sheet_id=sheet_id, + hlc_physical=hlc_p, + hlc_logical=hlc_l, + op_type=op_type, + raw_command=command, + payload=payload, + depends_on=depends_on or [], + ) + + def _classify(self, command: str) -> tuple[int, dict]: + """Classify a SocialCalc command and extract structured payload.""" + parts = command.split() + if not parts: + return OpType.SET_CELL_VALUE, {} + + verb = parts[0].lower() + + # "set A1 value 42" + # "set A1 formula =SUM(A1:A5)" + # "set A1 bgcolor #ff0000" + # "set A1:B5 bgcolor #ff0000" + if verb == "set" and len(parts) >= 4: + cell_or_range = parts[1] + attr = parts[2].lower() + value = " ".join(parts[3:]) + + if attr == "value": + return OpType.SET_CELL_VALUE, {"cell": cell_or_range, "value": value} + elif attr == "formula": + return OpType.SET_CELL_FORMULA, { + "cell": cell_or_range, + "formula": value, + } + else: + op = ( + OpType.SET_RANGE_FORMAT + if ":" in cell_or_range + else OpType.SET_CELL_FORMAT + ) + return op, {"cell": cell_or_range, "attr": attr, "value": value} + + # "erase A1" + if verb == "erase" and len(parts) >= 2: + return OpType.CLEAR_CELL, {"cell": parts[1]} + + # "insertrow 3 1" + if verb == "insertrow" and len(parts) >= 2: + return OpType.INSERT_ROW, { + "row": int(parts[1]), + "count": int(parts[2]) if len(parts) > 2 else 1, + } + + # "deleterow 3 1" + if verb == "deleterow" and len(parts) >= 2: + return OpType.DELETE_ROW, { + "row": int(parts[1]), + "count": int(parts[2]) if len(parts) > 2 else 1, + } + + # "insertcol C 1" + if verb == "insertcol" and len(parts) >= 2: + return OpType.INSERT_COL, { + "col": parts[1], + "count": int(parts[2]) if len(parts) > 2 else 1, + } + + # "deletecol C 1" + if verb == "deletecol" and len(parts) >= 2: + return OpType.DELETE_COL, { + "col": parts[1], + "count": int(parts[2]) if len(parts) > 2 else 1, + } + + # "sort A1:D10 A up B down" + if verb == "sort": + return OpType.SORT_SHEET, {"raw": command} + + # "movepaste A1:B5 C3 all" + if verb in ("movepaste", "moveinsert"): + return OpType.MOVE_RANGE, {"raw": command} + + # "copy A1:B5 C3 all" / "paste C3 all" + if verb in ("copy", "paste"): + return OpType.COPY_RANGE, {"raw": command} + + # Unknown — preserve as raw for forward-compatibility + return OpType.SET_CELL_VALUE, {"raw": command} + + def make_snapshot_request(self, sheet_id: str) -> Operation: + hlc_p, hlc_l = _clock.now() + return Operation( + op_id=str(uuid.uuid4()), + peer_id=self._peer_id, + sheet_id=sheet_id, + hlc_physical=hlc_p, + hlc_logical=hlc_l, + op_type=OpType.SNAPSHOT_REQUEST, + raw_command="", + payload={}, + ) + + def make_snapshot_chunk( + self, + sheet_id: str, + chunk_index: int, + total_chunks: int, + data: bytes, + op_log_from: int = 0, + ) -> Operation: + hlc_p, hlc_l = _clock.now() + return Operation( + op_id=str(uuid.uuid4()), + peer_id=self._peer_id, + sheet_id=sheet_id, + hlc_physical=hlc_p, + hlc_logical=hlc_l, + op_type=OpType.SNAPSHOT_CHUNK, + raw_command="", + payload={ + "idx": chunk_index, + "total": total_chunks, + "data": data.hex(), # bytes as hex string for msgpack compat + "op_log_from": op_log_from, + }, + ) + + def make_heartbeat(self, sheet_id: str, cell_cursor: str = "") -> Operation: + hlc_p, hlc_l = _clock.now() + return Operation( + op_id=str(uuid.uuid4()), + peer_id=self._peer_id, + sheet_id=sheet_id, + hlc_physical=hlc_p, + hlc_logical=hlc_l, + op_type=OpType.HEARTBEAT, + raw_command="", + payload={"cursor": cell_cursor}, + ) + + +# --------------------------------------------------------------------------- +# Receive-side: update HLC from incoming operation +# --------------------------------------------------------------------------- + + +def receive_operation(op: Operation) -> Operation: + """ + Update local HLC from an incoming operation's timestamp. + + Must be called for every received operation to maintain + causal consistency guarantees. + """ + _clock.update(op.hlc_physical, op.hlc_logical) + return op diff --git a/examples/p2pcalc/p2pcalc/p2p_node.py b/examples/p2pcalc/p2pcalc/p2p_node.py new file mode 100644 index 000000000..321a66c93 --- /dev/null +++ b/examples/p2pcalc/p2pcalc/p2p_node.py @@ -0,0 +1,425 @@ +# p2pcalc/p2p_node.py — py-libp2p node for P2PCalc +# Author: Dashpreet Singh +# +# Taps directly into the existing GossipSub v2.0 implementation in +# py-libp2p. No changes to the library — we are purely a consumer. +# +# Innovations beyond baseline: +# 1. Topic-per-sheet: each EtherCalc room gets its own GossipSub topic +# "p2pcalc/" — allows fine-grained mesh membership +# 2. Peer presence via HEARTBEAT operations — know who is editing what +# 3. Late-joiner state sync via libp2p streams (not GossipSub pub-sub) +# Snapshot is chunked and sent directly, then op-log replayed on top +# 4. mDNS for LAN discovery (zero config), DHT for WAN discovery + +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable +import json +import logging + +from multiaddr import Multiaddr + +from libp2p.abc import IHost, INetStream +from libp2p.crypto.secp256k1 import create_new_key_pair +from libp2p.custom_types import TProtocol +from libp2p.peer.peerinfo import info_from_p2p_addr +from libp2p.pubsub.gossipsub import GossipSub +from libp2p.pubsub.pb import rpc_pb2 +from libp2p.pubsub.pubsub import Pubsub + +from .crdt import ConflictPolicy, SheetCRDT +from .operation import Operation, OperationFactory, OpType, receive_operation + +logger = logging.getLogger("p2pcalc.node") + +# GossipSub topic prefix +TOPIC_PREFIX = "p2pcalc" + +# Protocol ID for direct state-sync streams +SYNC_PROTOCOL: TProtocol = TProtocol("/p2pcalc/sync/1.0.0") + +# GossipSub parameters tuned for collaborative editing +# Lower heartbeat = faster mesh convergence for small peer groups +GOSSIPSUB_PARAMS: dict[str, int | bool] = { + "degree": 6, + "degree_low": 4, + "degree_high": 8, + "heartbeat_interval": 1, # 1s heartbeat for fast convergence + "gossip_window": 3, + "gossip_history": 5, + "time_to_live": 60, + "adaptive_gossip_enabled": True, # GossipSub v2.0 adaptive features + "spam_protection_enabled": True, +} + + +def _topic(sheet_id: str) -> str: + return f"{TOPIC_PREFIX}/{sheet_id}" + + +class P2PCalcNode: + """ + A P2PCalc peer node. + + Lifecycle: + 1. __init__ — configure keys and sheet state + 2. start() — spin up libp2p host, GossipSub, mDNS + 3. join_sheet() — subscribe to sheet topic, request snapshot if needed + 4. publish() — broadcast an Operation to all peers on topic + 5. stop() — clean shutdown + + Incoming operations are delivered via the callback registered in + join_sheet(). The node handles dedup, HLC update, and CRDT apply + before calling the callback. + """ + + def __init__( + self, + port: int = 0, + known_peers: list[str] | None = None, + conflict_policy: ConflictPolicy = ConflictPolicy.MULTI_VALUE, + ): + self._port = port + self._known_peers = known_peers or [] + self._conflict_policy = conflict_policy + + # Keyed by sheet_id + self._sheets: dict[str, SheetCRDT] = {} + self._callbacks: dict[str, Callable[[Operation], Awaitable[None]]] = {} + # sheet_id -> peer_id -> cursor + self._presence: dict[str, dict[str, str]] = {} + + # Set during start() + self._host: IHost | None = None + self._pubsub: Pubsub | None = None + self._gossipsub: GossipSub | None = None + self._factory: OperationFactory | None = None + self._peer_id_str: str = "" + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _require_host(self) -> IHost: + """Return _host, raising RuntimeError if node has not been started.""" + if self._host is None: + raise RuntimeError("Node not started — call start() first") + return self._host + + def _require_pubsub(self) -> Pubsub: + """Return _pubsub, raising RuntimeError if node has not been started.""" + if self._pubsub is None: + raise RuntimeError("Node not started — call start() first") + return self._pubsub + + # ------------------------------------------------------------------ + # Startup + # ------------------------------------------------------------------ + + async def start(self) -> str: + """Start the libp2p host and GossipSub. Returns peer_id string.""" + key_pair = create_new_key_pair() + + listen_addr = Multiaddr(f"/ip4/0.0.0.0/tcp/{self._port}") + + from libp2p import new_host as _new_host + + self._host = _new_host(key_pair=key_pair) # type: ignore[assignment] + if self._host is None: + raise RuntimeError("new_host() returned None — cannot start node") + + await self._host.get_network().listen(listen_addr) + + self._peer_id_str = str(self._host.get_id()) + self._factory = OperationFactory(peer_id=self._peer_id_str) + + logger.info("P2PCalc node started: %s", self._peer_id_str) + + # Boot GossipSub v2.0 + from libp2p.pubsub.gossipsub import ( + PROTOCOL_ID, + PROTOCOL_ID_V11, + PROTOCOL_ID_V13, + PROTOCOL_ID_V20, + ) + + self._gossipsub = GossipSub( # pyrefly: ignore + protocols=[ + PROTOCOL_ID_V20, + PROTOCOL_ID_V13, + PROTOCOL_ID_V11, + PROTOCOL_ID, + ], + **GOSSIPSUB_PARAMS, # pyrefly: ignore + ) + self._pubsub = Pubsub(self._host, self._gossipsub) + + # Register direct sync stream handler for late joiners + self._host.set_stream_handler(SYNC_PROTOCOL, self._handle_sync_stream) + + # Connect to any known peers + for addr_str in self._known_peers: + await self._connect_peer(addr_str) + + return self._peer_id_str + + async def _connect_peer(self, addr_str: str) -> None: + host = self._require_host() + try: + peer_info = info_from_p2p_addr(Multiaddr(addr_str)) + await host.connect(peer_info) + logger.info("Connected to peer: %s", addr_str) + except Exception as e: + logger.warning("Failed to connect to %s: %s", addr_str, e) + + # ------------------------------------------------------------------ + # Sheet management + # ------------------------------------------------------------------ + + async def join_sheet( + self, + sheet_id: str, + on_operation: Callable[[Operation], Awaitable[None]], + request_snapshot: bool = True, + ) -> None: + """Subscribe to a sheet's GossipSub topic and request state sync.""" + pubsub = self._require_pubsub() + + if sheet_id not in self._sheets: + self._sheets[sheet_id] = SheetCRDT( + sheet_id=sheet_id, + policy=self._conflict_policy, + ) + + self._callbacks[sheet_id] = on_operation + topic = _topic(sheet_id) + + # Subscribe and register message handler + await pubsub.subscribe(topic) + pubsub.add_validator(topic, self._validate_message) # type: ignore[attr-defined] + + # Start message pump for this topic + asyncio.create_task(self._message_pump(sheet_id, topic)) + + logger.info("Joined sheet: %s (topic=%s)", sheet_id, topic) + + # Request snapshot from existing peers if we're a late joiner + if request_snapshot and self._factory: + snap_req = self._factory.make_snapshot_request(sheet_id) + await self.publish(snap_req) + + async def leave_sheet(self, sheet_id: str) -> None: + pubsub = self._require_pubsub() + topic = _topic(sheet_id) + await pubsub.unsubscribe(topic) + self._callbacks.pop(sheet_id, None) + logger.info("Left sheet: %s", sheet_id) + + # ------------------------------------------------------------------ + # Publish + # ------------------------------------------------------------------ + + async def publish(self, op: Operation) -> None: + """Broadcast an Operation to all peers on this sheet's topic.""" + pubsub = self._require_pubsub() + topic = _topic(op.sheet_id) + data = op.to_bytes() + await pubsub.publish(topic, data) + logger.debug("Published op %s type=%s", op.op_id[:8], op.op_type) + + async def publish_command( + self, + sheet_id: str, + socialcalc_command: str, + ) -> Operation: + """Parse a SocialCalc command string, apply locally, and broadcast.""" + if not self._factory: + raise RuntimeError("Node not started") + + sheet = self._sheets.get(sheet_id) + depends_on = [] + if sheet and sheet.op_log_length() > 0: + last_op = sheet.op_log_since(sheet.op_log_length() - 1) + if last_op: + depends_on = [last_op[0].op_id] + + op = self._factory.from_socialcalc_command( + socialcalc_command, sheet_id, depends_on=depends_on + ) + + # Apply locally first (optimistic) + if sheet: + sheet.apply(op) + + await self.publish(op) + return op + + # ------------------------------------------------------------------ + # Message handling + # ------------------------------------------------------------------ + + async def _message_pump(self, sheet_id: str, topic: str) -> None: + """Consume GossipSub messages for this sheet.""" + pubsub = self._require_pubsub() + subscription = pubsub.get_subscription(topic) # type: ignore[attr-defined] + if subscription is None: + logger.warning("No subscription found for topic %s", topic) + return + + async for msg in subscription: + if msg is None: + continue + await self._handle_message(sheet_id, msg) + + async def _handle_message(self, sheet_id: str, msg: rpc_pb2.Message) -> None: + try: + op = Operation.from_bytes(msg.data) + except Exception as e: + logger.warning("Failed to deserialise operation: %s", e) + return + + # Integrity check + if not op.verify_integrity(): + logger.warning("Integrity check failed for op %s", op.op_id[:8]) + return + + # Update HLC + receive_operation(op) + + # Skip own messages + if op.peer_id == self._peer_id_str: + return + + sheet = self._sheets.get(sheet_id) + if not sheet: + return + + # Handle control operations + if op.op_type == OpType.SNAPSHOT_REQUEST: + await self._serve_snapshot(op) + return + + if op.op_type == OpType.HEARTBEAT: + cursor = op.payload.get("cursor", "") + if sheet_id not in self._presence: + self._presence[sheet_id] = {} + self._presence[sheet_id][op.peer_id] = cursor + logger.debug("Peer %s cursor: %s", op.peer_id[:8], cursor) + return + + # Apply to CRDT + applied = sheet.apply(op) + if not applied: + logger.debug("Duplicate op %s — skipped", op.op_id[:8]) + return + + # Deliver to adapter callback + cb = self._callbacks.get(sheet_id) + if cb: + await cb(op) + + async def _validate_message(self, peer_id: str, msg: rpc_pb2.Message) -> bool: + """GossipSub message validator — reject malformed operations.""" + try: + op = Operation.from_bytes(msg.data) + return op.verify_integrity() + except Exception: + return False + + # ------------------------------------------------------------------ + # State sync for late joiners + # ------------------------------------------------------------------ + + async def _serve_snapshot(self, request_op: Operation) -> None: + """Respond to a SNAPSHOT_REQUEST by sending current state.""" + sheet_id = request_op.sheet_id + sheet = self._sheets.get(sheet_id) + if not sheet: + return + + snapshot_data = json.dumps(sheet.snapshot()).encode() + + # Chunk into 64KB pieces for libp2p stream compatibility + chunk_size = 64 * 1024 + chunks = [ + snapshot_data[i : i + chunk_size] + for i in range(0, len(snapshot_data), chunk_size) + ] + if not chunks: + chunks = [b"{}"] + + if not self._factory: + return + + for i, chunk in enumerate(chunks): + chunk_op = self._factory.make_snapshot_chunk( + sheet_id=sheet_id, + chunk_index=i, + total_chunks=len(chunks), + data=chunk, + op_log_from=sheet.op_log_length(), + ) + await self.publish(chunk_op) + + logger.info( + "Served snapshot for %s in %d chunks to %s", + sheet_id, + len(chunks), + request_op.peer_id[:8], + ) + + async def _handle_sync_stream(self, stream: INetStream) -> None: + """Handle direct stream requests (for large snapshots).""" + try: + data = await stream.read(4096) + sheet_id = data.decode().strip() + sheet = self._sheets.get(sheet_id) + if sheet: + payload = json.dumps(sheet.snapshot()).encode() + await stream.write(len(payload).to_bytes(4, "big") + payload) + await stream.close() + except Exception as e: + logger.error("Sync stream error: %s", e) + + # ------------------------------------------------------------------ + # Peer presence + # ------------------------------------------------------------------ + + async def send_heartbeat(self, sheet_id: str, cursor: str = "") -> None: + if not self._factory: + return + hb = self._factory.make_heartbeat(sheet_id, cursor) + await self.publish(hb) + + def get_presence(self, sheet_id: str) -> dict[str, str]: + """Return dict of peer_id -> cell cursor for a sheet.""" + return self._presence.get(sheet_id, {}) + + # ------------------------------------------------------------------ + # Introspection + # ------------------------------------------------------------------ + + @property + def peer_id(self) -> str: + return self._peer_id_str + + def get_listen_addrs(self) -> list[str]: + if not self._host: + return [] + return [ + str(addr) + "/p2p/" + self._peer_id_str for addr in self._host.get_addrs() + ] + + def get_sheet_state(self, sheet_id: str) -> SheetCRDT | None: + return self._sheets.get(sheet_id) + + def get_conflicts(self, sheet_id: str) -> list: + sheet = self._sheets.get(sheet_id) + return sheet.get_conflicts() if sheet else [] + + async def stop(self) -> None: + if self._host: + await self._host.close() + logger.info("P2PCalc node stopped") diff --git a/examples/p2pcalc/p2pcalc/requirements.txt b/examples/p2pcalc/p2pcalc/requirements.txt new file mode 100644 index 000000000..dfa7a5489 --- /dev/null +++ b/examples/p2pcalc/p2pcalc/requirements.txt @@ -0,0 +1,6 @@ +py-libp2p>=0.2.1 +msgpack>=1.0.7 +redis[asyncio]>=5.0.4 +multiaddr>=0.0.9 +pytest>=8.2.0 +pytest-asyncio>=0.23.6 diff --git a/examples/p2pcalc/p2pcalc/state_sync.py b/examples/p2pcalc/p2pcalc/state_sync.py new file mode 100644 index 000000000..33cc8c964 --- /dev/null +++ b/examples/p2pcalc/p2pcalc/state_sync.py @@ -0,0 +1,283 @@ +# p2pcalc/state_sync.py — Late-joiner state synchronisation +# Author: Dashpreet Singh +# +# INNOVATION: Two-phase state reconstruction +# +# Phase 1: Snapshot transfer +# The late joiner requests a snapshot via GossipSub SNAPSHOT_REQUEST. +# Any peer that has state responds with SNAPSHOT_CHUNK operations. +# Multiple peers may respond — we pick the first complete snapshot +# and verify it with a content hash. +# +# Phase 2: Op-log replay +# The snapshot includes op_log_from = N (the op count at snapshot time). +# Any operations that arrived after N are replayed on top. +# This is exactly how Git's pack-file + reflog system works. +# +# ADDITIONAL INNOVATION: Snapshot race resolution +# Multiple peers may send snapshots simultaneously. We use the peer +# with the highest op count as the authoritative source, since they +# have seen the most history. If counts are equal, lower peer_id wins. + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass, field +import hashlib +import json +import logging +from typing import IO + +from .crdt import SheetCRDT +from .operation import Operation + +logger = logging.getLogger("p2pcalc.sync") + +# How long to wait for snapshot chunks before timing out +SNAPSHOT_TIMEOUT_SECONDS = 15 + +# Maximum snapshot size we accept (10 MB) +MAX_SNAPSHOT_BYTES = 10 * 1024 * 1024 + + +@dataclass +class SnapshotCandidate: + """A snapshot being assembled from chunk operations.""" + + peer_id: str + total_chunks: int + op_log_from: int # replay ops from this index + chunks: dict[int, bytes] = field(default_factory=dict) + received_at: float = field(default_factory=lambda: __import__("time").time()) + + @property + def is_complete(self) -> bool: + return len(self.chunks) == self.total_chunks + + @property + def data(self) -> bytes: + return b"".join(self.chunks[i] for i in range(self.total_chunks)) + + @property + def content_hash(self) -> str: + return hashlib.sha256(self.data).hexdigest()[:16] + + +class StateSyncManager: + """ + Manages late-joiner state synchronisation for a single sheet. + + Used by P2PCalcNode internally when a new peer joins a sheet and + requests a snapshot. + """ + + def __init__(self, sheet: SheetCRDT): + self._sheet = sheet + self._candidates: dict[str, SnapshotCandidate] = {} + self._synced = False + self._sync_event = asyncio.Event() + + def handle_snapshot_chunk(self, op: Operation) -> bool: + """ + Process an incoming SNAPSHOT_CHUNK operation. + + Returns True when a complete, valid snapshot has been assembled + and applied. + """ + if self._synced: + return True + + payload = op.payload + peer_id = op.peer_id + chunk_idx = payload.get("idx", 0) + total = payload.get("total", 1) + data_hex = payload.get("data", "") + op_log_from = payload.get("op_log_from", 0) + + try: + chunk_data = bytes.fromhex(data_hex) + except ValueError: + logger.warning("Invalid chunk data from peer %s", peer_id[:8]) + return False + + # Get or create candidate for this peer + if peer_id not in self._candidates: + self._candidates[peer_id] = SnapshotCandidate( + peer_id=peer_id, + total_chunks=total, + op_log_from=op_log_from, + ) + + candidate = self._candidates[peer_id] + candidate.chunks[chunk_idx] = chunk_data + + if not candidate.is_complete: + logger.debug( + "Snapshot from %s: %d/%d chunks received", + peer_id[:8], + len(candidate.chunks), + total, + ) + return False + + return self._apply_candidate(candidate) + + def _apply_candidate(self, candidate: SnapshotCandidate) -> bool: + """Validate and apply a complete snapshot candidate.""" + data = candidate.data + + if len(data) > MAX_SNAPSHOT_BYTES: + logger.warning( + "Snapshot from %s too large (%d bytes) — rejected", + candidate.peer_id[:8], + len(data), + ) + return False + + try: + snapshot = json.loads(data.decode()) + except (json.JSONDecodeError, UnicodeDecodeError) as e: + logger.warning( + "Invalid snapshot JSON from %s: %s", candidate.peer_id[:8], e + ) + return False + + # If multiple peers sent snapshots, pick the one with most history + best = self._best_candidate() + if best and best.peer_id != candidate.peer_id: + if best.op_log_from > candidate.op_log_from: + logger.debug( + "Ignoring snapshot from %s — peer %s has more history", + candidate.peer_id[:8], + best.peer_id[:8], + ) + return False + + # Apply snapshot to sheet state + self._apply_snapshot_dict(snapshot) + self._synced = True + self._sync_event.set() + + logger.info( + "Snapshot applied from peer %s: %d cells, op_log_from=%d", + candidate.peer_id[:8], + len(snapshot.get("cells", {})), + candidate.op_log_from, + ) + return True + + def _best_candidate(self) -> SnapshotCandidate | None: + """Return the candidate with the most op history.""" + complete = [c for c in self._candidates.values() if c.is_complete] + if not complete: + return None + return max( + complete, + key=lambda c: (c.op_log_from, -ord(c.peer_id[0])), + ) + + def _apply_snapshot_dict(self, snapshot: dict) -> None: + """Reconstruct sheet cell state from a snapshot dict.""" + cells = snapshot.get("cells", {}) + + from .operation import OperationFactory + + factory = OperationFactory(peer_id="__snapshot__") + + for coord, cell_data in cells.items(): + value = cell_data.get("value", "") + formula = cell_data.get("formula", "") + + if formula: + command = f"set {coord} formula {formula}" + elif value: + command = f"set {coord} value {value}" + else: + continue + + op = factory.from_socialcalc_command(command, self._sheet.sheet_id) + self._sheet.apply(op) + + logger.debug("Applied %d cells from snapshot", len(cells)) + + async def wait_for_sync(self, timeout: float = SNAPSHOT_TIMEOUT_SECONDS) -> bool: + """Wait until a snapshot has been successfully applied.""" + try: + await asyncio.wait_for(self._sync_event.wait(), timeout=timeout) + return True + except asyncio.TimeoutError: + logger.warning( + "Snapshot sync timed out after %.1fs — starting from empty state", + timeout, + ) + self._synced = True + return False + + @property + def is_synced(self) -> bool: + return self._synced + + +class OpLogPersistence: + """ + Persists the operation log to local disk for crash recovery. + + INNOVATION: On restart, a peer replays its own op-log to recover + its last known state before requesting a network snapshot. This + means peers can recover from crashes without needing the network, + and reduces the burden on other peers to serve full snapshots. + + Format: newline-delimited MessagePack operations (like a WAL). + """ + + def __init__(self, sheet_id: str, data_dir: str = "./p2pcalc_data"): + from pathlib import Path + + _dir = Path(data_dir) + _dir.mkdir(parents=True, exist_ok=True) + self._path = str(_dir / f"{sheet_id}.oplog") + self._file: IO[bytes] | None = None + + async def open(self) -> None: + self._file = open(self._path, "ab") + logger.info("Op-log opened: %s", self._path) + + async def append(self, op: Operation) -> None: + if not self._file: + return + data = op.to_bytes() + # Write: 4-byte length prefix + msgpack bytes + import struct + + self._file.write(struct.pack(">I", len(data)) + data) + self._file.flush() + + def replay(self) -> list[Operation]: + """Read all operations from the log file.""" + ops = [] + import struct + + try: + with open(self._path, "rb") as f: + while True: + header = f.read(4) + if len(header) < 4: + break + length = struct.unpack(">I", header)[0] + data = f.read(length) + if len(data) < length: + break + try: + ops.append(Operation.from_bytes(data)) + except Exception as e: + logger.warning("Corrupt op-log entry: %s", e) + break + except FileNotFoundError: + pass + logger.info("Replayed %d operations from local op-log", len(ops)) + return ops + + async def close(self) -> None: + if self._file: + self._file.close() + self._file = None diff --git a/examples/p2pcalc/tests/__init__.py b/examples/p2pcalc/tests/__init__.py new file mode 100644 index 000000000..ed555e595 --- /dev/null +++ b/examples/p2pcalc/tests/__init__.py @@ -0,0 +1 @@ +"""P2PCalc test suite.""" diff --git a/examples/p2pcalc/tests/test_p2pcalc.py b/examples/p2pcalc/tests/test_p2pcalc.py new file mode 100644 index 000000000..91f99d1fa --- /dev/null +++ b/examples/p2pcalc/tests/test_p2pcalc.py @@ -0,0 +1,466 @@ +# tests/test_p2pcalc.py — Test suite for P2PCalc +# Author: Dashpreet Singh +# +# Run: python -m pytest tests/test_p2pcalc.py -v +# No network or Redis needed — all tests are pure logic. + +from pathlib import Path +import sys +import time +import unittest + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from p2pcalc.crdt import ( # pyrefly: ignore + ConflictPolicy, + MultiValueRegister, + SheetCRDT, + StructuralCRDT, +) +from p2pcalc.operation import ( # pyrefly: ignore + HLC, + Operation, + OperationFactory, + OpType, +) +from p2pcalc.state_sync import SnapshotCandidate # pyrefly: ignore + +# --------------------------------------------------------------------------- +# HLC Tests +# --------------------------------------------------------------------------- + + +class TestHLC(unittest.TestCase): + def test_monotonically_increasing(self): + clock = HLC() + timestamps = [clock.now() for _ in range(100)] + for i in range(1, len(timestamps)): + a, b = timestamps[i - 1], timestamps[i] + # b must be strictly greater than a + self.assertTrue( + b[0] > a[0] or (b[0] == a[0] and b[1] > a[1]), + f"HLC not monotonic: {a} -> {b}", + ) + + def test_update_advances_past_remote(self): + clock = HLC() + # Simulate a remote timestamp far in the future + future_physical = int(time.time() * 1000) + 5000 + clock.update(future_physical, 0) + ts = clock.now() + self.assertGreaterEqual(ts[0], future_physical) + + def test_compare_physical_dominates(self): + clock = HLC() + a = (1000, 5) + b = (2000, 0) + self.assertEqual(clock.compare(a, b), -1) + self.assertEqual(clock.compare(b, a), 1) + + def test_compare_logical_breaks_tie(self): + clock = HLC() + a = (1000, 3) + b = (1000, 7) + self.assertEqual(clock.compare(a, b), -1) + self.assertEqual(clock.compare(b, a), 1) + + def test_compare_equal(self): + clock = HLC() + a = (1000, 3) + self.assertEqual(clock.compare(a, a), 0) + + +# --------------------------------------------------------------------------- +# Operation Tests +# --------------------------------------------------------------------------- + + +class TestOperationFactory(unittest.TestCase): + def setUp(self): + self.factory = OperationFactory(peer_id="peer-A") + + def test_parse_set_cell_value(self): + op = self.factory.from_socialcalc_command("set A1 value 42", "sheet1") + self.assertEqual(op.op_type, OpType.SET_CELL_VALUE) + self.assertEqual(op.payload["cell"], "A1") + self.assertEqual(op.payload["value"], "42") + self.assertEqual(op.sheet_id, "sheet1") + self.assertEqual(op.peer_id, "peer-A") + + def test_parse_set_cell_formula(self): + op = self.factory.from_socialcalc_command( + "set B2 formula =SUM(A1:A5)", "sheet1" + ) + self.assertEqual(op.op_type, OpType.SET_CELL_FORMULA) + self.assertEqual(op.payload["cell"], "B2") + self.assertEqual(op.payload["formula"], "=SUM(A1:A5)") + + def test_parse_set_cell_format(self): + op = self.factory.from_socialcalc_command("set C3 bgcolor #ff0000", "sheet1") + self.assertEqual(op.op_type, OpType.SET_CELL_FORMAT) + self.assertEqual(op.payload["attr"], "bgcolor") + + def test_parse_erase(self): + op = self.factory.from_socialcalc_command("erase A1", "sheet1") + self.assertEqual(op.op_type, OpType.CLEAR_CELL) + self.assertEqual(op.payload["cell"], "A1") + + def test_parse_insert_row(self): + op = self.factory.from_socialcalc_command("insertrow 3 1", "sheet1") + self.assertEqual(op.op_type, OpType.INSERT_ROW) + self.assertEqual(op.payload["row"], 3) + self.assertEqual(op.payload["count"], 1) + + def test_parse_delete_col(self): + op = self.factory.from_socialcalc_command("deletecol C 2", "sheet1") + self.assertEqual(op.op_type, OpType.DELETE_COL) + self.assertEqual(op.payload["col"], "C") + + def test_parse_range_format(self): + op = self.factory.from_socialcalc_command("set A1:B5 bgcolor blue", "sheet1") + self.assertEqual(op.op_type, OpType.SET_RANGE_FORMAT) + + def test_serialisation_roundtrip(self): + op = self.factory.from_socialcalc_command("set A1 value hello", "sheet1") + restored = Operation.from_bytes(op.to_bytes()) + self.assertEqual(restored.op_id, op.op_id) + self.assertEqual(restored.peer_id, op.peer_id) + self.assertEqual(restored.raw_command, op.raw_command) + self.assertEqual(restored.payload, op.payload) + self.assertEqual(restored.hlc, op.hlc) + + def test_integrity_check_passes(self): + op = self.factory.from_socialcalc_command("set A1 value 99", "sheet1") + self.assertTrue(op.verify_integrity()) + + def test_integrity_check_fails_on_tamper(self): + op = self.factory.from_socialcalc_command("set A1 value 99", "sheet1") + op.raw_command = "set A1 value 999" # tampered + self.assertFalse(op.verify_integrity()) + + def test_snapshot_request(self): + op = self.factory.make_snapshot_request("sheet1") + self.assertEqual(op.op_type, OpType.SNAPSHOT_REQUEST) + + def test_snapshot_chunk(self): + op = self.factory.make_snapshot_chunk( + "sheet1", chunk_index=0, total_chunks=2, data=b"hello", op_log_from=42 + ) + self.assertEqual(op.op_type, OpType.SNAPSHOT_CHUNK) + self.assertEqual(op.payload["idx"], 0) + self.assertEqual(op.payload["total"], 2) + self.assertEqual(op.payload["op_log_from"], 42) + + def test_heartbeat(self): + op = self.factory.make_heartbeat("sheet1", cell_cursor="B7") + self.assertEqual(op.op_type, OpType.HEARTBEAT) + self.assertEqual(op.payload["cursor"], "B7") + + def test_hlc_assigned(self): + op1 = self.factory.from_socialcalc_command("set A1 value 1", "s") + op2 = self.factory.from_socialcalc_command("set A2 value 2", "s") + t1 = op1.hlc + t2 = op2.hlc + self.assertTrue(t2[0] > t1[0] or (t2[0] == t1[0] and t2[1] > t1[1])) + + def test_depends_on_propagated(self): + op = self.factory.from_socialcalc_command( + "set A1 value x", "s", depends_on=["abc123"] + ) + self.assertIn("abc123", op.depends_on) + + +# --------------------------------------------------------------------------- +# CRDT — MultiValueRegister Tests +# --------------------------------------------------------------------------- + + +class TestMultiValueRegister(unittest.TestCase): + def _make_op(self, peer_id, cell, value, hlc_p, hlc_l): + factory = OperationFactory(peer_id=peer_id) + op = factory.from_socialcalc_command(f"set {cell} value {value}", "s") + op.hlc_physical = hlc_p + op.hlc_logical = hlc_l + op.payload = {"cell": cell, "value": value} + return op + + def test_lww_higher_hlc_wins(self): + mvr = MultiValueRegister(policy=ConflictPolicy.LAST_WRITE_WINS) + op1 = self._make_op("peer-A", "A1", "first", 1000, 0) + op2 = self._make_op("peer-B", "A1", "second", 2000, 0) + mvr.apply(op1) + mvr.apply(op2) + state = mvr.get_cell("A1") + self.assertEqual(state.current_value, "second") + self.assertFalse(state.is_conflict) + + def test_lww_lower_hlc_ignored(self): + mvr = MultiValueRegister(policy=ConflictPolicy.LAST_WRITE_WINS) + op1 = self._make_op("peer-A", "A1", "first", 2000, 0) + op2 = self._make_op("peer-B", "A1", "stale", 1000, 0) + mvr.apply(op1) + mvr.apply(op2) + state = mvr.get_cell("A1") + self.assertEqual(state.current_value, "first") + + def test_mvr_concurrent_writes_produce_conflict(self): + mvr = MultiValueRegister(policy=ConflictPolicy.MULTI_VALUE) + # Same HLC = concurrent writes + op1 = self._make_op("peer-A", "A1", "alice_value", 1000, 0) + op2 = self._make_op("peer-B", "A1", "bob_value", 1000, 0) + mvr.apply(op1) + mvr.apply(op2) + state = mvr.get_cell("A1") + self.assertTrue(state.is_conflict) + self.assertEqual(len(state.candidates), 2) + + def test_mvr_causal_write_resolves_conflict(self): + mvr = MultiValueRegister(policy=ConflictPolicy.MULTI_VALUE) + op1 = self._make_op("peer-A", "A1", "alice", 1000, 0) + op2 = self._make_op("peer-B", "A1", "bob", 1000, 0) + # op3 causally follows both (higher HLC) — should dominate + op3 = self._make_op("peer-C", "A1", "resolved", 2000, 0) + mvr.apply(op1) + mvr.apply(op2) + mvr.apply(op3) + state = mvr.get_cell("A1") + self.assertFalse(state.is_conflict) + self.assertEqual(state.current_value, "resolved") + + def test_mvr_clear_removes_all_candidates(self): + mvr = MultiValueRegister(policy=ConflictPolicy.MULTI_VALUE) + op1 = self._make_op("peer-A", "A1", "value", 1000, 0) + mvr.apply(op1) + factory = OperationFactory(peer_id="peer-A") + erase = factory.from_socialcalc_command("erase A1", "s") + erase.payload = {"cell": "A1"} + mvr.apply(erase) + state = mvr.get_cell("A1") + self.assertEqual(len(state.candidates), 0) + + def test_peer_priority_lower_peer_wins(self): + mvr = MultiValueRegister(policy=ConflictPolicy.PEER_PRIORITY) + op1 = self._make_op("peer-A", "A1", "alice", 1000, 0) + op2 = self._make_op("peer-B", "A1", "bob", 1000, 0) + mvr.apply(op2) + mvr.apply(op1) + state = mvr.get_cell("A1") + # "peer-A" < "peer-B" lexicographically + self.assertEqual(state.current_value, "alice") + + def test_get_conflicts_empty_initially(self): + mvr = MultiValueRegister() + self.assertEqual(mvr.get_conflicts(), []) + + def test_snapshot_export(self): + mvr = MultiValueRegister(policy=ConflictPolicy.LAST_WRITE_WINS) + op = self._make_op("peer-A", "B3", "test_val", 1000, 0) + mvr.apply(op) + snap = mvr.snapshot() + self.assertIn("B3", snap) + self.assertEqual(snap["B3"]["value"], "test_val") + + +# --------------------------------------------------------------------------- +# CRDT — StructuralCRDT Tests +# --------------------------------------------------------------------------- + + +class TestStructuralCRDT(unittest.TestCase): + def _make_row_op(self, op_type, row, peer_id="peer-A", hlc=(1000, 0)): + factory = OperationFactory(peer_id=peer_id) + op = factory.from_socialcalc_command(f"insertrow {row} 1", "s") + op.op_type = op_type + op.payload = {"row": row, "count": 1} + op.hlc_physical = hlc[0] + op.hlc_logical = hlc[1] + return op + + def test_insert_row_appears_in_visible(self): + crdt = StructuralCRDT() + op = self._make_row_op(OpType.INSERT_ROW, 3) + crdt.apply(op) + self.assertIn(3, crdt.visible_rows()) + + def test_delete_row_tombstoned(self): + crdt = StructuralCRDT() + op_ins = self._make_row_op(OpType.INSERT_ROW, 5) + op_del = self._make_row_op(OpType.DELETE_ROW, 5) + crdt.apply(op_ins) + crdt.apply(op_del) + self.assertNotIn(5, crdt.visible_rows()) + + def test_concurrent_inserts_both_appear(self): + crdt = StructuralCRDT() + op_a = self._make_row_op(OpType.INSERT_ROW, 3, "peer-A") + op_b = self._make_row_op(OpType.INSERT_ROW, 3, "peer-B") + crdt.apply(op_a) + crdt.apply(op_b) + rows = crdt.visible_rows() + self.assertEqual(rows.count(3), 2) + + def test_delete_of_nonexistent_is_noop(self): + crdt = StructuralCRDT() + op = self._make_row_op(OpType.DELETE_ROW, 99) + crdt.apply(op) # should not raise + self.assertNotIn(99, crdt.visible_rows()) + + def test_snapshot_structure(self): + crdt = StructuralCRDT() + crdt.apply(self._make_row_op(OpType.INSERT_ROW, 1)) + snap = crdt.snapshot() + self.assertIn("rows", snap) + self.assertIn("cols", snap) + + +# --------------------------------------------------------------------------- +# SheetCRDT Tests +# --------------------------------------------------------------------------- + + +class TestSheetCRDT(unittest.TestCase): + def test_apply_and_dedup(self): + sheet = SheetCRDT("test-sheet") + factory = OperationFactory("peer-X") + op = factory.from_socialcalc_command("set A1 value 10", "test-sheet") + result1 = sheet.apply(op) + result2 = sheet.apply(op) # duplicate + self.assertTrue(result1) + self.assertFalse(result2) + + def test_op_log_grows(self): + sheet = SheetCRDT("test-sheet") + factory = OperationFactory("peer-X") + for i in range(5): + op = factory.from_socialcalc_command( + f"set A{i + 1} value {i}", "test-sheet" + ) + sheet.apply(op) + self.assertEqual(sheet.op_log_length(), 5) + + def test_op_log_since(self): + sheet = SheetCRDT("test-sheet") + factory = OperationFactory("peer-X") + for i in range(5): + op = factory.from_socialcalc_command( + f"set A{i + 1} value {i}", "test-sheet" + ) + sheet.apply(op) + recent = sheet.op_log_since(3) + self.assertEqual(len(recent), 2) + + def test_snapshot_roundtrip(self): + sheet = SheetCRDT("test-sheet") + factory = OperationFactory("peer-X") + op = factory.from_socialcalc_command("set B5 value hello", "test-sheet") + sheet.apply(op) + snap = sheet.snapshot() + self.assertEqual(snap["sheet_id"], "test-sheet") + self.assertIn("cells", snap) + self.assertIn("B5", snap["cells"]) + + +# --------------------------------------------------------------------------- +# Snapshot Candidate Tests +# --------------------------------------------------------------------------- + + +class TestSnapshotCandidate(unittest.TestCase): + def test_completeness_detection(self): + candidate = SnapshotCandidate(peer_id="peer-A", total_chunks=3, op_log_from=10) + self.assertFalse(candidate.is_complete) + candidate.chunks[0] = b"chunk0" + candidate.chunks[1] = b"chunk1" + candidate.chunks[2] = b"chunk2" + self.assertTrue(candidate.is_complete) + + def test_data_assembly(self): + candidate = SnapshotCandidate(peer_id="peer-A", total_chunks=2, op_log_from=0) + candidate.chunks[0] = b"hello " + candidate.chunks[1] = b"world" + self.assertEqual(candidate.data, b"hello world") + + def test_content_hash_deterministic(self): + candidate = SnapshotCandidate(peer_id="peer-A", total_chunks=1, op_log_from=0) + candidate.chunks[0] = b"test data" + h1 = candidate.content_hash + h2 = candidate.content_hash + self.assertEqual(h1, h2) + + +# --------------------------------------------------------------------------- +# Integration: multi-peer convergence simulation +# --------------------------------------------------------------------------- + + +class TestConvergence(unittest.TestCase): + """Simulate two peers editing concurrently and verify convergence.""" + + def test_two_peers_same_cell_lww_converges(self): + sheet_a = SheetCRDT("shared", policy=ConflictPolicy.LAST_WRITE_WINS) + sheet_b = SheetCRDT("shared", policy=ConflictPolicy.LAST_WRITE_WINS) + + factory_a = OperationFactory("peer-A") + factory_b = OperationFactory("peer-B") + + # Peer A edits A1 at t=1000 + op_a = factory_a.from_socialcalc_command("set A1 value from_A", "shared") + op_a.hlc_physical = 1000 + op_a.hlc_logical = 0 + op_a.payload = {"cell": "A1", "value": "from_A"} + + # Peer B edits A1 at t=2000 (later, causally after) + op_b = factory_b.from_socialcalc_command("set A1 value from_B", "shared") + op_b.hlc_physical = 2000 + op_b.hlc_logical = 0 + op_b.payload = {"cell": "A1", "value": "from_B"} + + # Both peers receive both ops (in different orders) + sheet_a.apply(op_a) + sheet_a.apply(op_b) + + sheet_b.apply(op_b) + sheet_b.apply(op_a) + + val_a = sheet_a.cells.get_cell("A1").current_value + val_b = sheet_b.cells.get_cell("A1").current_value + + # Both converge to the same value (op_b wins due to higher HLC) + self.assertEqual(val_a, val_b) + self.assertEqual(val_a, "from_B") + + def test_three_peers_distinct_cells_no_conflict(self): + sheets = [ + SheetCRDT("shared", policy=ConflictPolicy.MULTI_VALUE) for _ in range(3) + ] + factories = [OperationFactory(f"peer-{i}") for i in range(3)] + + ops = [] + for i, (factory, sheet) in enumerate(zip(factories, sheets)): + op = factory.from_socialcalc_command( + f"set {chr(65 + i)}1 value peer_{i}", "shared" + ) + op.payload = {"cell": f"{chr(65 + i)}1", "value": f"peer_{i}"} + ops.append(op) + + # All peers receive all ops + for sheet in sheets: + for op in ops: + sheet.apply(op) + + # No conflicts — each peer edited a different cell + for sheet in sheets: + self.assertEqual(sheet.get_conflicts(), []) + + # All sheets have same state + for sheet in sheets: + snap = sheet.snapshot() + for i in range(3): + coord = f"{chr(65 + i)}1" + self.assertIn(coord, snap["cells"]) + + +if __name__ == "__main__": + from unittest import main as _main + + _main(verbosity=2) diff --git a/pyproject.toml b/pyproject.toml index 585e91cfe..f28e2b11f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -95,6 +95,7 @@ dev = [ { include-group = "docs" }, { include-group = "test" }, { include-group = "release" }, + { include-group = "p2pcalc" }, "ipython", "mypy>=1.15.0", "pre-commit>=3.4.0", @@ -130,6 +131,10 @@ release = [ "twine", "wheel", ] +p2pcalc = [ + "msgpack>=1.0.7", + "redis[asyncio]>=5.0.4", +] [tool.setuptools] include-package-data = true