diff --git a/Cargo.toml b/Cargo.toml index 17e51a9..8677e94 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,5 +30,36 @@ thiserror = "2.0" unsafe_code = "forbid" [workspace.lints.clippy] -all = "warn" -pedantic = "warn" +bool_comparison = "deny" +duplicate_mod = "deny" +inconsistent_struct_constructor = "deny" +invalid_regex = "deny" +mem_forget = "deny" +mixed_case_hex_literals = "deny" +suspicious_arithmetic_impl = "deny" +uninit_assumed_init = "deny" +suspicious_else_formatting = "deny" +suspicious_op_assign_impl = "deny" +suspicious_to_owned = "deny" +cmp_owned = "deny" +cmp_null = "deny" +manual_map = "deny" + +too_many_arguments = "warn" +cognitive_complexity = "warn" +large_enum_variant = "warn" +needless_borrow = "warn" +needless_pass_by_value = "warn" +redundant_clone = "warn" +unnecessary_cast = "warn" +inefficient_to_string = "warn" +or_fun_call = "warn" +unnecessary_to_owned = "warn" +map_clone = "warn" +flat_map_identity = "warn" +needless_collect = "warn" +vec_init_then_push = "warn" + +len_zero = "allow" +range_plus_one = "allow" +manual_range_contains = "allow" diff --git a/ROADMAP.md b/ROADMAP.md index 8503724..f82a6f1 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -1,6 +1,6 @@ # Inkfinite vNext / Version 2 -Status: implementation in progress; V2-01 through V2-18 are complete +Status: implementation in progress; V2-01 through V2-19 are complete This is the product and architecture contract for vNext. [TODO.md](TODO.md) is the implementation queue. @@ -29,12 +29,15 @@ the CLI and a bundled `SKILL.md`; MCP and UI automation are not part of vNext. to v2-native fixtures. - The desktop app, file-mode CLI, and live CLI use the same engine. Stale or invalid mutations cannot partly modify a document. +- A user and a fresh Codex session can co-design a wireframe in the offline + desktop app through inspect, query, proposal review, revision, partial + acceptance, save, reopen, validation, and deterministic rendering. - Pages contain ordered layers. Hidden and locked layers affect rendering and interaction; new shapes and built-in stencils use the active layer. - A scripted 10,000-shape document meets the recorded interaction budget on the project reference machine. Benchmarks decide whether a spatial index ships. -- Shared fixtures cover Rust/TypeScript schema agreement, migrations, CRDT - merges, transactions, rendering, CLI output, and IPC. +- Shared fixtures cover Rust/TypeScript schema agreement, CRDT merges, + transactions, rendering, CLI output, and IPC. ## Current state @@ -63,9 +66,9 @@ the CLI and a bundled `SKILL.md`; MCP and UI automation are not part of vNext. - The desktop owns an authenticated, versioned local IPC server. The CLI can inspect open sessions, query the same materialized records as file mode, and request frontend focus without a TCP listener or background daemon. -- [TODO.md](TODO.md) starts the remaining work at V2-19: offline replica sync, - followed by agent packaging, release verification, and v1 compatibility - removal. +- [TODO.md](TODO.md) starts the remaining work at V2-20: agent packaging, + compatibility removal, collaborative desktop QA, and final release + verification. V2-19's offline replica sync is complete. ## Architecture @@ -154,13 +157,13 @@ rules where specified, and validated before the session adopts them. The canonical v2 file is Automerge's compact binary form with the `.inkfinite` extension. `.inkfinite.json` is a stable snapshot export, not a CRDT round-trip -format. Through V2-21, the development build also accepts the frozen v1 envelope -as an import source so the architecture and release baselines remain -reproducible. V2-22 removes that unreleased compatibility surface before vNext -ships. [docs/v2-file-format.md](docs/v2-file-format.md) defines the current file -behavior. The CLI supplies JSON inspection for repositories and CI. The Rust -core produces deterministic SVG, and V2-16 exposed that renderer through the -CLI. +format. The development build still accepts the frozen v1 envelope so the +architecture baselines remain reproducible. V2-22 replaces useful predecessor +coverage with native fixtures and removes that unreleased compatibility surface +before collaborative QA and the final release matrix. +[docs/v2-file-format.md](docs/v2-file-format.md) defines the current file +behavior. The CLI supplies JSON inspection for repositories and CI and produces +deterministic SVG. ## Document contract @@ -250,9 +253,10 @@ actor-scoped change and must preserve concurrent work from other actors. ## Desktop and renderer The Tauri backend maintains open document sessions with path, Automerge state, -materialized snapshot, undo/redo metadata, sync peers, and dirty/recovery state. -Commands cover create/open/snapshot/commit/undo/redo/save/save-as/query/validate/ -close and proposal accept/reject. +materialized snapshot, undo/redo metadata, trusted sync peers, and +dirty/recovery state. Commands cover create/open/snapshot/commit/undo/redo/save/ +save-as/query/validate, peer connect/disconnect/send/receive, close, and +proposal accept/reject. Use commands for request/response, events for small notifications, and channels only for ordered streams. The frontend applies returned patches to its mirror; @@ -278,29 +282,10 @@ spatial index; WebGL and OffscreenCanvas remain deferred pending evidence. ## CLI and agent workflow The CLI binary is `inkfinite`; the desktop application is `Inkfinite`. File mode -works without the app. The shipped commands are: - -```sh -inkfinite new architecture.inkfinite -inkfinite inspect architecture.inkfinite --json -inkfinite query architecture.inkfinite --role architecture.service --json -inkfinite validate architecture.inkfinite -inkfinite schema document -inkfinite capabilities --json -inkfinite app status --json -``` - -V2-16 added mutating and rendering commands: - -```sh -inkfinite apply architecture.inkfinite --transaction transaction.json --dry-run -inkfinite render architecture.inkfinite --output architecture.svg -``` - -Structured `shape create/patch/delete`, `connect`, and `layout` commands build -ordinary transactions and never bypass the engine. `schema document`, -`schema transaction`, `schema protocol`, and `capabilities --json` expose the -contracts. +works without the app. It can create, inspect, query, validate, mutate, and render +closed files. Structured shape, connection, and layout commands build ordinary +transactions and never bypass the engine. Schema commands and +`capabilities --json` expose the machine-readable contracts. `--json` and `--non-interactive` are global and may appear before or after a subcommand. Every command supports deterministic JSON where applicable, reads @@ -309,16 +294,9 @@ stderr, never prompts under either global option, and uses stable exit codes. Mutations support `--dry-run`, report heads and affected IDs, write atomically, and refuse failed preconditions or invalid documents. -File-mode commands return 0 for success, 2 for invalid usage, 3 for file or -input errors, 4 for invalid documents or data, and 5 for file, lock, or state -conflicts. `inkfinite --help` and `inkfinite capabilities --json` expose the -same contract. - CLI design follows the [Command Line Interface Guidelines](https://clig.dev/). -Top-level help shows common examples and links to documentation and issue -reporting. Each subcommand has realistic examples and clear value names. New -commands must keep help, capabilities, README.md, TODO.md, ROADMAP.md, and -integration tests synchronized. +`inkfinite --help`, `inkfinite capabilities --json`, README.md, and integration +tests are the command and exit-code references and must stay synchronized. The bundled skill teaches agents to inspect heads, query narrowly, create the smallest transaction, dry-run, resolve conflicts, apply or propose, validate, @@ -326,6 +304,13 @@ and render affected content. Agents must respect locked shapes and `agent_editable: false`, prefer semantic selectors and layout operations, and never edit document bytes manually. +The release-candidate QA pairs a user with a fresh Codex session to wireframe a +desktop application while disconnected from the network. Codex works through +the live proposal path; the user reviews, rejects, revises, and partially accepts +changes while continuing to edit in the canvas. Save, reopen, undo/redo, +validation, and SVG rendering must preserve the reviewed result. This is a human +visual review boundary, not desktop UI automation. + ## Live control and collaboration The CLI connects to the running desktop app for `app status`, `app inspect`, @@ -348,10 +333,12 @@ one-time authorization grant. The server does not expose public TCP or local HTTP. The Tauri process owns the server and removes its discovery record when it exits; vNext has no background daemon. -Automerge sync between trusted peers is a vNext deliverable. The sync layer must -be transport-independent and prove offline concurrent edits between two app -instances. A hosted relay, accounts, sharing policy, and presence service are -later milestones; CRDT storage and merge semantics are not deferred with them. +Automerge sync between trusted peers is implemented through a transport-neutral +envelope and bounded per-peer checkpoints. The session validates and repairs a +fork before adopting it, quarantines malformed payloads, and persists document +state before advancing a durable checkpoint. A hosted relay, accounts, sharing +policy, and presence service remain later milestones; CRDT storage and merge +semantics are local and complete. ## Persistence and generated bindings @@ -407,25 +394,19 @@ pnpm --filter @inkfinite/web lint ``` V2-21 extends this with generated-artifact, CLI, IPC, convergence, recovery, -performance, accessibility, and visual checks. Human review remains required for -stencils, render parity, proposal UX, recovery prompts, and permission failures. - -## Milestones - -Milestones 1 through 6 are complete: the architecture gate, Rust authority, -desktop vertical slice, editor scale work, layers, opacity, stencils, -deterministic SVG, and the mutating file-mode CLI all passed. -[TODO.md](TODO.md) owns ticket-level status and acceptance criteria. - -- **Milestone 5, layers and styles:** complete. -- **Milestone 6, CLI and SVG:** complete. The file-mode CLI exposes the SVG - renderer and validated, atomic mutation commands. -- **Milestone 7, live control and sync:** authenticated IPC, reviewable - proposals, explicit apply, and one-time authorization are complete; two-replica - offline convergence remains. -- **Milestone 8, agent and release readiness:** bundle the agent skill, record - release evidence, replace useful v1 coverage with v2-native fixtures, remove - the unreleased v1 compatibility surface, and rerun the release matrix. +performance, accessibility, and visual checks after native-model cleanup and the +collaborative desktop QA pass. Human review remains required for stencils, +render parity, proposal UX, recovery prompts, and permission failures. + +## Remaining sequence + +Milestones 1 through 7 are complete. [TODO.md](TODO.md) owns ticket-level status, +acceptance criteria, and dependency order. The remaining path is: + +1. Bundle and verify the agent skill (V2-20). +2. Replace useful predecessor coverage and collapse to the native model (V2-22). +3. Run the human-and-Codex desktop wireframing sequence (V2-23). +4. Run the final release matrix against that native candidate (V2-21). ## Deferred milestones @@ -442,9 +423,9 @@ deterministic SVG, and the mutating file-mode CLI all passed. ## Boundaries -- Follow existing patterns, preserve v1 fixtures through V2-21, replace useful - coverage before V2-22 removes them, run affected tests, and keep Git read-only - unless repository instructions change. +- Follow existing patterns, preserve predecessor fixtures until V2-22 replaces + their useful coverage, run affected tests, and keep Git read-only unless + repository instructions change. - Ask before adding production dependencies, changing a published format or protocol, broadening authentication or network exposure, or deleting user data. - Never expose a public control server, accept invalid partial writes, hand-edit diff --git a/TODO.md b/TODO.md index 41ef158..f71c75f 100644 --- a/TODO.md +++ b/TODO.md @@ -2,6 +2,9 @@ These tickets implement [ROADMAP.md](ROADMAP.md). +Tickets appear in dependency order. Their IDs stay stable when the sequence +changes. + ## Milestone 1: Architecture gate Exit when the v1 baseline is reproducible and the CRDT decision has measured @@ -179,31 +182,41 @@ and integration/accessibility coverage. What to build: Wire the selected CRDT sync protocol through a transport-neutral peer layer and prove desktop-to-desktop convergence. +Implemented: Added a trusted, transport-neutral Automerge peer envelope with +bounded retry and quarantine state. Document files persist per-peer checkpoints +beside the recovery data, and the session and Tauri boundaries expose connect, +disconnect, send, and receive operations without adding a hosted service. + Blocked by: V2-07, V2-08, V2-17 Acceptance criteria: -- [ ] Two trusted app instances exchange only missing changes, reconnect after +- [x] Two trusted app instances exchange only missing changes, reconnect after offline edits, and converge to the same materialized snapshot and heads. -- [ ] Concurrent shape edits, hierarchy moves, layer changes, text edits, +- [x] Concurrent shape edits, hierarchy moves, layer changes, text edits, deletions, and undo trigger the specified merge or deterministic repair. -- [ ] Duplicate, delayed, and retried messages are harmless; corrupt or invalid +- [x] Duplicate, delayed, and retried messages are harmless; corrupt or invalid merged state is quarantined without replacing the open valid session. -- [ ] Peer state, actor IDs, compaction, and recovery survive restart without +- [x] Peer state, actor IDs, compaction, and recovery survive restart without resending unbounded history. -- [ ] The transport boundary does not assume hosted accounts, a public relay, or +- [x] The transport boundary does not assume hosted accounts, a public relay, or durable presence. Verification: -- Run fixed-seed and property-based two-replica tests with reordered, duplicated, - delayed, disconnected, restarted, and corrupted message cases. +- Run fixed-schedule and property-based two-replica tests with reordered, + duplicated, delayed, disconnected, restarted, and corrupted message cases. + +Verification completed with Rust peer-envelope, quarantine, reorder, and +property tests; session-level offline/restart tests; and desktop TypeScript +contract and adapter tests. ## Milestone 8: Agent and release readiness -Exit when a clean installation completes every human, CLI, proposal, recovery, -and sync acceptance path, useful predecessor coverage has native replacements, -and no unreleased predecessor model remains in the product or codebase. +Exit when a clean installation supports a complete human-and-agent wireframing +session, every CLI, proposal, recovery, and sync acceptance path passes, useful +predecessor coverage has native replacements, and no unreleased predecessor +model remains in the product or codebase. ### V2-20: Bundle the Inkfinite agent skill @@ -229,52 +242,6 @@ Verification: - Install the packaged skill in a clean test environment and run all examples against fixture copies. -### V2-21: Run the vNext release matrix - -What to build: Assemble release evidence and close every roadmap success -criterion without adding new architecture. - -Blocked by: V2-11, V2-12, V2-13, V2-14, V2-16, V2-18, V2-19, V2-20 - -Acceptance criteria: - -- [ ] All Rust, TypeScript, browser, shared-fixture, generated-artifact, CLI, IPC, - convergence, migration, recovery, accessibility, and visual tests pass. -- [ ] The recorded 10,000-shape frame, query, memory, open, save, and sync budgets - pass on reference hardware with results attached. -- [ ] A corpus of real v1 documents imports without data loss; rollback and - backup restoration are rehearsed. -- [ ] Human review signs off layers, proposal UX, render parity, recovery prompts, - and permission failures on supported desktop platforms. -- [ ] File-format documentation accurately describes the behavior under - evaluation; V2-22 collapses the temporary predecessor/current split before - release. - -Verification: - -```sh -cargo fmt --all -- --check -cargo test --workspace --all-features -cargo clippy --workspace --all-targets --all-features -- -D warnings -pnpm format:check -pnpm bindings:check -pnpm bindings:test -pnpm --filter @inkfinite/core test --run -pnpm --filter @inkfinite/renderer test --run -pnpm --filter @inkfinite/runtime typecheck -pnpm --filter @inkfinite/input-dom typecheck -pnpm --filter @inkfinite/ui test -pnpm --filter @inkfinite/web test -pnpm --filter @inkfinite/desktop test -pnpm --filter @inkfinite/bindings typecheck -pnpm --filter @inkfinite/core typecheck -pnpm --filter @inkfinite/renderer typecheck -pnpm --filter @inkfinite/ui check -pnpm --filter @inkfinite/web check -pnpm --filter @inkfinite/web lint -pnpm --filter @inkfinite/desktop check -``` - ### V2-22: Collapse to one native Inkfinite model What to build: Treat the current document model and file flow as Inkfinite's @@ -285,7 +252,7 @@ documentation. Keep explicit file, schema, and protocol version fields where they validate persisted or exchanged data or allow future evolution; they must not preserve an implementation of the unreleased predecessor. -Blocked by: V2-21 +Blocked by: V2-20 Acceptance criteria: @@ -303,6 +270,10 @@ Acceptance criteria: recovery, and import/export coverage with native fixtures before deleting its predecessor source. The replacement tests must exercise the same observable failure and recovery cases. +- [ ] Before removing the import path, run the real predecessor-document corpus + once, record the results, and convert representative successful, invalid, + and recovery cases into native fixtures. Keep backup and rollback evidence + with the release records rather than retaining compatibility code. - [ ] Update ROADMAP.md and current documentation to describe the native model and supported file flow directly, without presenting the codebase as a compatibility bridge between product generations. @@ -319,6 +290,109 @@ Verification: current documentation for predecessor types and transition-oriented terms; review every match rather than relying on a fixed list of file removals. +### V2-23: Complete a human-and-Codex desktop wireframing session + +What to build: Run and record a realistic QA session in which a user and Codex +co-design a desktop application wireframe through the packaged Inkfinite skill, +the live CLI, and the desktop proposal review surface. + +Blocked by: V2-20, V2-22 + +Acceptance criteria: + +- [ ] Start from a clean desktop build and a fresh Codex context with only the + packaged skill installed. After launch, disconnect the network; the rest + of the session requires no account, server, browser automation, raw + document edit, or unbundled repository instruction. +- [ ] The user creates or selects a local `.inkfinite` document and gives Codex + a short desktop-application brief. The target wireframe exercises named + semantic roles, text, connections, at least two layers, and one locked or + `agent_editable: false` element. +- [ ] Codex discovers the open session, inspects its heads, queries only the + relevant records, and describes a small first change before proposing it. + The first durable agent action uses `app propose`, not direct apply. +- [ ] The user rejects one proposal and confirms that the snapshot and heads do + not change, then requests a revision and reviews its ghost preview and + created, changed, and deleted IDs. +- [ ] The user partially accepts a proposal whose operations can remain valid + independently. Inkfinite commits only the selected operations once and + Codex re-inspects the result before continuing. +- [ ] The user makes an intervening canvas edit while a later proposal is open. + Accepting the stale proposal produces a refreshed review or a clear + conflict; it never commits silently against old heads or loses the human + edit. +- [ ] Codex completes the wireframe with reviewed proposals while respecting + locked layers, `agent_editable`, and semantic selectors. The user can also + continue editing with normal desktop tools between proposals. +- [ ] Save and close the document while still offline. File-mode `validate` and + deterministic SVG rendering pass; after reopen, the canvas and + `app inspect` agree with those artifacts and undo/redo remains coherent + across accepted work. +- [ ] Record the brief, Inkfinite and skill versions, platform, commands, review + decisions, final `.inkfinite` file, SVG, screenshots, and any defects. A + failed step gets a focused follow-up ticket and the sequence is rerun after + its fix. + +Verification: + +- Build and launch the desktop release candidate, install the V2-20 skill in a + clean Codex environment, disable networking, and complete the sequence above + without UI automation. Attach the command transcript and artifacts to the + release evidence. +- Use [tldraw offline](https://tldraw.dev/blog/tldraw-offline) only as a product + frame of reference: local files and trusted local agents. Inkfinite agents must + continue through validated transactions and authenticated local IPC; they do + not execute arbitrary code in the canvas or replace an open file externally. + +### V2-21: Run the vNext release matrix + +What to build: Assemble release evidence and close every roadmap success +criterion against the native release candidate without adding new architecture. + +Blocked by: V2-11, V2-12, V2-13, V2-14, V2-16, V2-18, V2-19, V2-20, V2-22, +V2-23 + +Acceptance criteria: + +- [ ] All Rust, TypeScript, browser, shared-fixture, generated-artifact, CLI, IPC, + convergence, recovery, accessibility, and visual tests pass against the + native model. +- [ ] The recorded 10,000-shape frame, query, memory, open, save, and sync budgets + pass on reference hardware with results attached. +- [ ] Backup restoration and rollback are rehearsed with native files; malformed, + interrupted, and conflicting writes preserve the last valid document. +- [ ] Human review signs off layers, proposal UX, the V2-23 collaborative + wireframe, render parity, recovery prompts, and permission failures on + supported desktop platforms. +- [ ] File-format and release documentation describe exactly the native model + and behavior under evaluation. + +Verification: + +```sh +cargo fmt --all -- --check +cargo test --workspace --all-features +cargo clippy --workspace --all-targets --all-features -- -D warnings +pnpm format:check +pnpm bindings:check +pnpm bindings:test +pnpm --filter @inkfinite/core test --run +pnpm --filter @inkfinite/renderer test --run +pnpm --filter @inkfinite/runtime typecheck +pnpm --filter @inkfinite/input-dom typecheck +pnpm --filter @inkfinite/ui test +pnpm --filter @inkfinite/web test +pnpm --filter @inkfinite/desktop test +pnpm --filter @inkfinite/bindings typecheck +pnpm --filter @inkfinite/core typecheck +pnpm --filter @inkfinite/renderer typecheck +pnpm --filter @inkfinite/ui check +pnpm --filter @inkfinite/web check +pnpm --filter @inkfinite/web lint +pnpm --filter @inkfinite/desktop check +``` + ## Frontier -V2-19 is the current frontier. Sync two offline replicas and prove convergence. +V2-20 is the current frontier. Bundle the Inkfinite agent skill and prove its +examples in a clean environment. diff --git a/apps/desktop/src-tauri/src/ipc.rs b/apps/desktop/src-tauri/src/ipc.rs index a7c71d9..b145fab 100644 --- a/apps/desktop/src-tauri/src/ipc.rs +++ b/apps/desktop/src-tauri/src/ipc.rs @@ -19,6 +19,8 @@ pub const PROPOSAL_EVENT: &str = "inkfinite-proposal"; pub const PROPOSAL_CLEARED_EVENT: &str = "inkfinite-proposal-cleared"; /// Tauri event carrying a commit made by a live client. pub const COMMIT_EVENT: &str = "inkfinite-live-commit"; +/// Tauri event carrying a remote synchronization result. +pub const SYNC_EVENT: &str = "inkfinite-sync"; /// Handle used by the Tauri lifecycle to stop the local server and clean up discovery. pub struct IpcServerHandle { diff --git a/apps/desktop/src-tauri/src/lib.rs b/apps/desktop/src-tauri/src/lib.rs index bc6149a..03cc79b 100644 --- a/apps/desktop/src-tauri/src/lib.rs +++ b/apps/desktop/src-tauri/src/lib.rs @@ -32,6 +32,10 @@ pub fn run() { session::save_as, session::query, session::validate, + session::sync_connect, + session::sync_disconnect, + session::sync_next, + session::sync_receive, session::close, files::read_directory, files::rename_file, diff --git a/apps/desktop/src-tauri/src/session.rs b/apps/desktop/src-tauri/src/session.rs index be30ebd..34ce84f 100644 --- a/apps/desktop/src-tauri/src/session.rs +++ b/apps/desktop/src-tauri/src/session.rs @@ -7,8 +7,9 @@ use inkfinite_core::proto::{ TransactionDraft, }; use inkfinite_core::session::{ - SessionCommit, SessionError, SessionOpened, SessionSaved, SessionService, SessionStatus, + SessionCommit, SessionError, SessionOpened, SessionSaved, SessionService, SessionStatus, SessionSync, }; +use inkfinite_core::sync::SyncMessage; use inkfinite_core::{ActorId, ChangeHash, DocumentId}; use serde_json::json; use tauri::{AppHandle, Emitter, State}; @@ -195,6 +196,48 @@ pub fn validate(state: State<'_, DesktopState>, session_id: String) -> Result, session_id: String, peer_id: String) -> Result { + lock_service(&state)? + .connect_peer(&SessionId(session_id), ActorId::new(peer_id)) + .map_err(to_protocol_error) +} + +/// Removes a trusted peer checkpoint for one open document session. +#[tauri::command] +pub fn sync_disconnect(state: State<'_, DesktopState>, session_id: String, peer_id: String) -> Result { + let peer_id = ActorId::new(peer_id); + lock_service(&state)? + .disconnect_peer(&SessionId(session_id), &peer_id) + .map_err(to_protocol_error) +} + +/// Returns the next transport-neutral message for a connected peer. +#[tauri::command] +pub fn sync_next(state: State<'_, DesktopState>, session_id: String, peer_id: String) -> Result> { + lock_service(&state)? + .next_sync_message(&SessionId(session_id), &ActorId::new(peer_id)) + .map_err(to_protocol_error) +} + +/// Receives a transport-neutral message and returns the refreshed session +/// mirror after validated merge or quarantine. +#[tauri::command] +pub fn sync_receive( + app: AppHandle, state: State<'_, DesktopState>, session_id: String, message: SyncMessage, +) -> Result { + let session_id = SessionId(session_id); + let result = lock_service(&state)?.receive_sync_message(&session_id, &message); + if let Ok(update) = &result { + let _ = app.emit( + super::ipc::SYNC_EVENT, + json!({ "session_id": session_id, "sync": update }), + ); + } + result.map_err(to_protocol_error) +} + /// Closes a session and releases its advisory file lock. #[tauri::command] pub fn close(state: State<'_, DesktopState>, session_id: String) -> Result<()> { diff --git a/apps/desktop/src/lib/persistence/desktop-session.test.ts b/apps/desktop/src/lib/persistence/desktop-session.test.ts index d1ccacd..e95123c 100644 --- a/apps/desktop/src/lib/persistence/desktop-session.test.ts +++ b/apps/desktop/src/lib/persistence/desktop-session.test.ts @@ -249,6 +249,42 @@ function createFakeSessionApi() { return statusFor(args.session_id, session); }, + async syncConnect(args: Parameters[0]) { + const session = sessions.get(args.session_id); + if (!session) throw new Error("Missing fake session"); + session.status = { + ...session.status, + sync: { + status: "enabled", + peers: [ + { + peer_id: args.peer_id, + pending_messages: 0, + shared_heads: [], + quarantine: null, + }, + ], + warning: null, + }, + }; + return statusFor(args.session_id, session); + }, + + async syncDisconnect(args: Parameters[0]) { + const session = sessions.get(args.session_id); + if (!session) throw new Error("Missing fake session"); + session.status = { ...session.status, sync: { status: "disabled" } }; + return statusFor(args.session_id, session); + }, + + async syncNext(_args: Parameters[0]) { + return null; + }, + + async syncReceive(_args: Parameters[0]): Promise { + throw new Error("Sync receive is not part of this fake session"); + }, + async close(args: Parameters[0]) { sessions.delete(args.session_id); }, diff --git a/apps/desktop/src/lib/persistence/desktop-session.ts b/apps/desktop/src/lib/persistence/desktop-session.ts index 335ccb3..23b1ee7 100644 --- a/apps/desktop/src/lib/persistence/desktop-session.ts +++ b/apps/desktop/src/lib/persistence/desktop-session.ts @@ -47,7 +47,48 @@ export type SessionStatus = { recovery_available: boolean; can_undo: boolean; can_redo: boolean; - sync: { status: 'disabled' }; + sync: SyncState; +}; + +/** Trusted peer checkpoint projected across the desktop command boundary. */ +export type SyncPeerStatus = { + peer_id: string; + pending_messages: number; + shared_heads: ChangeHash[]; + quarantine: SyncQuarantine | null; +}; + +/** Diagnostic retained when a peer payload is rejected without replacing the document. */ +export type SyncQuarantine = { sequence: number; reason: string }; + +/** Session-level synchronization state. */ +export type SyncState = + | { status: 'disabled' } + | { status: 'enabled'; peers: SyncPeerStatus[]; warning: string | null }; + +/** Transport-neutral Automerge envelope exchanged by trusted peers. */ +export type SyncMessage = { + protocol_id: string; + version: number; + document_id: string; + sender: string; + recipient: string; + sequence: number; + payload: number[]; +}; + +/** Classification returned after one peer envelope is processed. */ +export type SyncDisposition = 'applied' | 'duplicate' | 'deferred' | 'quarantined'; + +/** Document patch and validation result returned by a peer merge. */ +export type SyncApplyResult = { + disposition: SyncDisposition; + adopted_messages: number; + heads: ChangeHash[]; + patch: CommitResult['patch']; + affected_ids: CommitResult['affected_ids']; + affected_regions: CommitResult['affected_regions']; + warnings: CommitResult['warnings']; }; /** Result returned after creating or opening a desktop session. */ @@ -59,6 +100,9 @@ export type SessionCommit = { commit: CommitResult; status: SessionStatus }; /** Result returned after persisting a session. */ export type SessionSaved = { save: { path: string; heads: ChangeHash[] }; status: SessionStatus }; +/** Result returned after adopting, deferring, or quarantining a peer message. */ +export type SessionSync = { sync: SyncApplyResult; status: SessionStatus }; + /** State update emitted when a live proposal is created, refreshed, or cleared. */ export type ProposalUpdate = { proposal: Proposal | null; message?: string }; @@ -87,6 +131,10 @@ export interface SessionApi { saveAs(args: { session_id: string; path: string; expected_heads: ChangeHash[] }): Promise; query(args: { session_id: string; query: Query }): Promise; validate(args: { session_id: string }): Promise; + syncConnect(args: { session_id: string; peer_id: string }): Promise; + syncDisconnect(args: { session_id: string; peer_id: string }): Promise; + syncNext(args: { session_id: string; peer_id: string }): Promise; + syncReceive(args: { session_id: string; message: SyncMessage }): Promise; close(args: { session_id: string }): Promise; } @@ -106,6 +154,10 @@ function createSessionApi(): SessionApi { saveAs: (args) => invoke('save_as', args), query: (args) => invoke('query', args), validate: (args) => invoke('validate', args), + syncConnect: (args) => invoke('sync_connect', args), + syncDisconnect: (args) => invoke('sync_disconnect', args), + syncNext: (args) => invoke('sync_next', args), + syncReceive: (args) => invoke('sync_receive', args), close: (args) => invoke('close', args) }; } @@ -128,6 +180,10 @@ export type DesktopSessionRepo = PersistentDocRepo & { acceptProposal(proposalId: string, operationPositions?: number[]): Promise; rejectProposal(proposalId: string): Promise; authorizeApply(): Promise; + syncConnect(peerId: string): Promise; + syncDisconnect(peerId: string): Promise; + syncNext(peerId: string): Promise; + syncReceive(message: SyncMessage): Promise; closeSession(): Promise; }; @@ -150,6 +206,7 @@ export function createDesktopSessionRepo(fileOps: DesktopFileOps, opts: { api?: type ProposalEvent = { session_id?: string | null; proposal: Proposal }; type ProposalClearedEvent = { message?: string }; type LiveCommitEvent = { session_id?: string | null; commit: SessionCommit }; + type LiveSyncEvent = { session_id?: string | null; sync: SessionSync }; function notifyProposal(update: ProposalUpdate) { currentProposal = update.proposal; @@ -183,6 +240,12 @@ export function createDesktopSessionRepo(fileOps: DesktopFileOps, opts: { api?: }) .then((stop) => liveUnlisteners.push(stop)) .catch(() => undefined); + void listen('inkfinite-sync', (event) => { + if (!eventBelongsToCurrentSession(event.payload.session_id)) return; + updateStatus(event.payload.sync.status); + }) + .then((stop) => liveUnlisteners.push(stop)) + .catch(() => undefined); } startLiveListeners(); @@ -484,6 +547,32 @@ export function createDesktopSessionRepo(fileOps: DesktopFileOps, opts: { api?: return api.authorizeApply({ session_id: currentStatus.session_id }); } + async function syncConnect(peerId: string): Promise { + if (!currentStatus) throw new Error('No board loaded'); + const status = await api.syncConnect({ session_id: currentStatus.session_id, peer_id: peerId }); + updateStatus(status); + return status; + } + + async function syncDisconnect(peerId: string): Promise { + if (!currentStatus) throw new Error('No board loaded'); + const status = await api.syncDisconnect({ session_id: currentStatus.session_id, peer_id: peerId }); + updateStatus(status); + return status; + } + + async function syncNext(peerId: string): Promise { + if (!currentStatus) throw new Error('No board loaded'); + return api.syncNext({ session_id: currentStatus.session_id, peer_id: peerId }); + } + + async function syncReceive(message: SyncMessage): Promise { + if (!currentStatus) throw new Error('No board loaded'); + const result = await api.syncReceive({ session_id: currentStatus.session_id, message }); + updateStatus(result.status); + return result; + } + return { kind: 'desktop', listBoards, @@ -510,6 +599,10 @@ export function createDesktopSessionRepo(fileOps: DesktopFileOps, opts: { api?: acceptProposal, rejectProposal, authorizeApply, + syncConnect, + syncDisconnect, + syncNext, + syncReceive, closeSession: closeCurrentSession }; } diff --git a/apps/desktop/src/lib/persistence/desktop-workspace.test.ts b/apps/desktop/src/lib/persistence/desktop-workspace.test.ts index 8dbd098..b290428 100644 --- a/apps/desktop/src/lib/persistence/desktop-workspace.test.ts +++ b/apps/desktop/src/lib/persistence/desktop-workspace.test.ts @@ -60,6 +60,10 @@ describe("desktop workspace adapter", () => { saveAs: async () => { throw new Error("not used"); }, query: async () => { throw new Error("not used"); }, validate: async () => { throw new Error("not used"); }, + syncConnect: async () => { throw new Error("not used"); }, + syncDisconnect: async () => { throw new Error("not used"); }, + syncNext: async () => { throw new Error("not used"); }, + syncReceive: async () => { throw new Error("not used"); }, close: async () => {}, }, }); diff --git a/crates/inkfinite-core/src/crdt/mod.rs b/crates/inkfinite-core/src/crdt/mod.rs index 7220f7d..1a6031b 100644 --- a/crates/inkfinite-core/src/crdt/mod.rs +++ b/crates/inkfinite-core/src/crdt/mod.rs @@ -67,7 +67,7 @@ pub struct MergeOutcome { pub enum CrdtError { /// Automerge rejected a document operation or encoded change. #[error("Automerge operation failed: {0}")] - Automerge(#[from] automerge::AutomergeError), + Automerge(#[source] Box), /// A typed snapshot could not be converted to or from its CRDT projection. #[error("document projection failed: {0}")] Projection(#[from] serde_json::Error), @@ -82,6 +82,12 @@ pub enum CrdtError { EmptyChange, } +impl From for CrdtError { + fn from(error: automerge::AutomergeError) -> Self { + Self::Automerge(Box::new(error)) + } +} + #[derive(Serialize, Deserialize)] struct StoredSnapshot { format: FormatId, @@ -313,6 +319,7 @@ pub trait SyncSession { } /// Per-peer Automerge synchronization state. +#[derive(Clone)] pub struct AutomergeSyncSession { state: AutomergeSyncState, } @@ -323,6 +330,50 @@ impl AutomergeSyncSession { pub fn new() -> Self { Self { state: AutomergeSyncState::new() } } + + /// Restores the compact state that Automerge documents for a peer. + /// + /// The encoded form intentionally contains only durable shared heads; + /// in-flight message bookkeeping is rebuilt on the next connection. + /// + /// # Errors + /// + /// Returns [`CrdtError::InvalidPayload`] when the checkpoint is malformed. + pub fn from_encoded_state(bytes: &[u8]) -> Result { + let state = AutomergeSyncState::decode(bytes).map_err(|error| CrdtError::InvalidPayload(error.to_string()))?; + Ok(Self { state }) + } + + /// Returns the bounded durable checkpoint for this peer. + #[must_use] + pub fn encoded_state(&self) -> Vec { + self.state.encode() + } + + /// Returns the causal heads known to be shared with the peer. + #[must_use] + pub fn shared_heads(&self) -> Vec { + hashes(self.state.shared_heads.clone()) + } + + /// Receives a sync payload and reports whether its newly advertised + /// changes became materialized rather than remaining as Automerge orphans. + pub(crate) fn receive_message_with_status( + &mut self, document: &mut AutomergeDocument, bytes: &[u8], + ) -> Result { + let message = + automerge::sync::Message::decode(bytes).map_err(|error| CrdtError::InvalidPayload(error.to_string()))?; + let before_heads = document.document.get_heads(); + let has_unavailable_head = + !message.changes.is_empty() && message.heads.iter().any(|head| !before_heads.contains(head)); + document + .document + .sync() + .receive_sync_message(&mut self.state, message)?; + let after_heads = document.document.get_heads(); + let document_advanced = after_heads.iter().any(|head| !before_heads.contains(head)); + Ok(!has_unavailable_head || document_advanced) + } } impl Default for AutomergeSyncSession { @@ -343,12 +394,7 @@ impl SyncSession for AutomergeSyncSession { } fn receive_message(&mut self, document: &mut AutomergeDocument, message: &[u8]) -> Result<(), Self::Error> { - let message = - automerge::sync::Message::decode(message).map_err(|error| CrdtError::InvalidPayload(error.to_string()))?; - document - .document - .sync() - .receive_sync_message(&mut self.state, message)?; + self.receive_message_with_status(document, message)?; Ok(()) } } diff --git a/crates/inkfinite-core/src/engine/error.rs b/crates/inkfinite-core/src/engine/error.rs index f5ba562..e99ef9c 100644 --- a/crates/inkfinite-core/src/engine/error.rs +++ b/crates/inkfinite-core/src/engine/error.rs @@ -1,4 +1,5 @@ use super::{ActorId, CrdtError, Error}; +use crate::sync::SyncError; /// Recoverable rejection from transaction, merge, validation, or query processing. #[derive(Debug, Error)] @@ -6,6 +7,9 @@ pub enum EngineError { /// The CRDT adapter could not complete an operation. #[error(transparent)] Crdt(#[from] CrdtError), + /// The peer synchronization boundary rejected an envelope or checkpoint. + #[error(transparent)] + Sync(#[from] SyncError), /// The transaction's serialized contract is structurally invalid. #[error("schema validation failed: {0}")] Schema(String), diff --git a/crates/inkfinite-core/src/engine/mod.rs b/crates/inkfinite-core/src/engine/mod.rs index a1dd035..b15d253 100644 --- a/crates/inkfinite-core/src/engine/mod.rs +++ b/crates/inkfinite-core/src/engine/mod.rs @@ -5,11 +5,14 @@ use std::cmp::Ordering; use std::collections::{BTreeMap, BTreeSet}; +use serde::{Deserialize, Serialize}; + use crate::crdt::{AutomergeDocument, CrdtError, EncodedChange}; use crate::proto::{ AffectedRegion, AssetPatch, Bounds, DocumentPatch, InverseMetadata, LayerContentsDisposition, LayerPatch, LayoutAxis, Operation, Query, QueryResult, RecordId, ShapeAlignment, ShapePatch, TransactionId, Warning, }; +use crate::sync::{PeerSync, SyncDisposition, SyncMessage}; use crate::{ ActorId, AssetId, BindingId, ChangeHash, ContainerLayout, Document, DocumentId, LayerId, Origin, PageId, RecordVersion, ShapeId, ShapeParent, ShapeProperties, ShapeRecord, SiblingAnchor, @@ -55,6 +58,26 @@ pub struct TransactionPreview { pub affected_regions: Vec, } +/// Materialized result of one peer synchronization exchange. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct SyncApplyResult { + /// Classification of the received message. + pub disposition: SyncDisposition, + /// Number of Automerge protocol messages adopted while draining pending + /// delayed messages. + pub adopted_messages: usize, + /// Causal heads after adoption or quarantine. + pub heads: Vec, + /// Records changed by remote edits or deterministic repair. + pub patch: DocumentPatch, + /// Records affected directly or by hierarchy repair. + pub affected_ids: Vec, + /// Document regions that should be redrawn by a frontend mirror. + pub affected_regions: Vec, + /// Deterministic merge repairs performed before adoption. + pub warnings: Vec, +} + struct PreparedTransaction { document: Document, inverse: Vec, @@ -64,6 +87,7 @@ struct PreparedTransaction { } /// Rust-owned document state plus actor-scoped undo and redo metadata. +#[derive(Clone)] pub struct TransactionEngine { crdt: AutomergeDocument, undo: BTreeMap>, @@ -272,6 +296,64 @@ impl TransactionEngine { Ok(warnings) } + /// Produces the next transport-neutral message for one connected peer. + /// + /// # Errors + /// + /// Returns a typed synchronization or CRDT error when the message cannot + /// be generated. + pub fn next_sync_message(&mut self, peer: &mut PeerSync) -> Result, EngineError> { + Ok(peer.next_message(&mut self.crdt)?) + } + + /// Receives one peer message on a fork, repairs deterministic hierarchy + /// damage, validates the materialized result, and adopts the fork only + /// after all checks pass. + /// + /// Corrupt payloads are quarantined by the peer checkpoint and return a + /// `Quarantined` disposition; the current document remains untouched. + /// Delayed messages remain queued until their missing dependencies arrive. + /// + /// # Errors + /// + /// Returns a typed envelope, synchronization, or invariant error when the + /// message cannot be addressed or the merged candidate cannot be repaired. + pub fn receive_sync_message( + &mut self, peer: &mut PeerSync, message: &SyncMessage, + ) -> Result { + let before = self.crdt.snapshot()?; + let mut candidate = self.crdt.clone(); + let mut candidate_peer = peer.clone(); + let peer_result = candidate_peer.receive_message(&mut candidate, message)?; + let mut candidate_snapshot = candidate.snapshot()?; + let mut warnings = Vec::new(); + + if validate_document(&candidate_snapshot.document).is_err() { + let original = candidate_snapshot.document.clone(); + warnings = repair_document(&mut candidate_snapshot.document)?; + validate_document(&candidate_snapshot.document)?; + if candidate_snapshot.document != original { + candidate.commit_document(&candidate_snapshot.document, "deterministic sync repair")?; + candidate_snapshot = candidate.snapshot()?; + } + } + validate_document(&candidate_snapshot.document)?; + + let (patch, affected_ids) = diff_documents(&before.document, &candidate_snapshot.document); + let affected_regions = affected_regions(&before.document, &candidate_snapshot.document, &affected_ids); + self.crdt = candidate; + *peer = candidate_peer; + Ok(SyncApplyResult { + disposition: peer_result.disposition, + adopted_messages: peer_result.applied_messages, + heads: candidate_snapshot.heads, + patch, + affected_ids, + affected_regions, + warnings, + }) + } + /// Returns changes after a caller's inspected heads. /// /// # Errors diff --git a/crates/inkfinite-core/src/engine/operations.rs b/crates/inkfinite-core/src/engine/operations.rs index 1caea2d..6d8c769 100644 --- a/crates/inkfinite-core/src/engine/operations.rs +++ b/crates/inkfinite-core/src/engine/operations.rs @@ -300,7 +300,7 @@ pub fn delete_page( .flat_map(|layer_id| descendant_ids_for_layer(document, layer_id)) .collect(); let mut inverse = vec![Operation::CreatePage { - page: crate::PageRecord { layer_ids: Vec::new(), version: RecordVersion(1), ..page.clone() }, + page: crate::PageRecord { layer_ids: Vec::new(), version: RecordVersion(1), ..page }, anchor, }]; for layer_id in &layer_ids { diff --git a/crates/inkfinite-core/src/file/mod.rs b/crates/inkfinite-core/src/file/mod.rs index 8ccfee6..7798c38 100644 --- a/crates/inkfinite-core/src/file/mod.rs +++ b/crates/inkfinite-core/src/file/mod.rs @@ -10,6 +10,7 @@ use std::path::PathBuf; use crate::engine::EngineError; +use crate::sync::SyncError; use thiserror::Error; /// Recoverable failure at the import, persistence, or recovery boundary. @@ -45,6 +46,9 @@ pub enum FileError { /// The transaction engine rejected a document or CRDT operation. #[error(transparent)] Engine(#[from] EngineError), + /// A trusted peer synchronization operation was rejected. + #[error(transparent)] + Sync(#[from] SyncError), /// A filesystem operation failed. #[error("{operation} {path}: {source}")] Io { @@ -70,7 +74,7 @@ pub use crate::engine::{CommitResult, TransactionDraft, TransactionEngine}; pub use migration::{ImportedV1, import_v1_json, parse_v1_json}; pub use persistence::{ DocumentFile, PersistenceOptions, SaveResult, export_snapshot_json, import_v1_file, import_v1_file_with_options, - read_v1_file, recovery_path_for, write_snapshot_json, + read_v1_file, recovery_path_for, sync_state_path_for, write_snapshot_json, }; #[cfg(test)] diff --git a/crates/inkfinite-core/src/file/persistence.rs b/crates/inkfinite-core/src/file/persistence.rs index b710e60..b8f8d3f 100644 --- a/crates/inkfinite-core/src/file/persistence.rs +++ b/crates/inkfinite-core/src/file/persistence.rs @@ -1,5 +1,6 @@ //! Canonical file persistence, advisory locks, atomic replacement, and recovery. +use std::collections::{BTreeMap, btree_map::Entry}; use std::fmt::Write as _; use std::fs::{self, File, OpenOptions}; use std::io::Write; @@ -7,7 +8,9 @@ use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; use crate::crdt::EncodedChange; -use crate::engine::{TransactionDraft, TransactionEngine}; +use crate::engine::{SyncApplyResult, TransactionDraft, TransactionEngine}; +use crate::ipc::CommitResult; +use crate::sync::{MAX_SYNC_PEERS, PeerSync, PeerSyncStatus, PersistedPeerSync, SyncError, SyncMessage, SyncStatus}; use crate::{ActorId, ChangeHash, Document, DocumentId, DocumentSnapshot}; use serde::{Deserialize, Serialize}; @@ -15,6 +18,9 @@ use super::{FileError, ImportedV1, import_v1_json}; const RECOVERY_FORMAT: &str = "inkfinite.recovery"; const RECOVERY_VERSION: u32 = 1; +const SYNC_STATE_FORMAT: &str = "inkfinite.sync-state"; +const SYNC_STATE_VERSION: u32 = 1; +const MAX_SYNC_STATE_BYTES: usize = 16 * 1024 * 1024; const DEFAULT_MAX_JOURNAL_ENTRIES: usize = 32; const DEFAULT_MAX_JOURNAL_BYTES: usize = 8 * 1024 * 1024; static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0); @@ -96,6 +102,17 @@ struct RecoveryFile { journal: Vec, } +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct SyncStateFile { + format: String, + version: u32, + source_path: String, + document_id: DocumentId, + actor_id: ActorId, + peers: Vec, +} + impl RecoveryFile { fn new(source_path: &Path, document_id: DocumentId, snapshot: Vec, heads: Vec) -> Self { Self { @@ -120,6 +137,8 @@ pub struct DocumentFile { baseline_bytes: Vec, baseline_heads: Vec, pending_recovery: Option, + sync_peers: BTreeMap, + sync_state_warning: Option, lock: AdvisoryLock, } @@ -150,7 +169,7 @@ impl DocumentFile { let bytes = read_bytes(&path, "read canonical document")?; let mut engine = load_canonical_bytes(&bytes, actor_id.clone())?; let heads = engine.snapshot()?.heads; - Ok(Self { + let mut session = Self { path, actor_id, engine, @@ -158,8 +177,12 @@ impl DocumentFile { baseline_bytes: bytes, baseline_heads: heads, pending_recovery: None, + sync_peers: BTreeMap::new(), + sync_state_warning: None, lock, - }) + }; + session.load_sync_state(); + Ok(session) } /// Creates and safely persists a new canonical document. @@ -194,9 +217,20 @@ impl DocumentFile { let mut engine = TransactionEngine::create(document_id, actor_id.clone(), document)?; let baseline_bytes = engine.save()?; let baseline_heads = engine.snapshot()?.heads; - let mut session = - Self { path, actor_id, engine, options, baseline_bytes, baseline_heads, pending_recovery: None, lock }; + let mut session = Self { + path, + actor_id, + engine, + options, + baseline_bytes, + baseline_heads, + pending_recovery: None, + sync_peers: BTreeMap::new(), + sync_state_warning: None, + lock, + }; session.save()?; + session.load_sync_state(); Ok(session) } @@ -258,7 +292,7 @@ impl DocumentFile { "recovery journal did not produce the recorded heads".into(), )); } - Ok(Self { + let mut session = Self { path, actor_id, engine, @@ -266,8 +300,12 @@ impl DocumentFile { baseline_bytes: recovery.snapshot.clone(), baseline_heads: recovery.base_heads.clone(), pending_recovery: Some(recovery), + sync_peers: BTreeMap::new(), + sync_state_warning: None, lock, - }) + }; + session.load_sync_state(); + Ok(session) } /// Returns the canonical path held by this session. @@ -327,12 +365,224 @@ impl DocumentFile { &mut self.engine } + /// Returns the bounded synchronization status for this document. + #[must_use] + pub fn sync_status(&self) -> SyncStatus { + SyncStatus { + peers: self.sync_peers.values().map(PeerSync::status).collect(), + warning: self.sync_state_warning.clone(), + } + } + + /// Trusts a peer actor and creates or returns its durable checkpoint. + /// + /// # Errors + /// + /// Returns a typed peer or filesystem error when the checkpoint cannot be + /// safely persisted. + pub fn connect_peer(&mut self, peer_id: ActorId) -> Result { + if let Some(peer) = self.sync_peers.get(&peer_id) { + return Ok(peer.status()); + } + if self.sync_peers.len() >= MAX_SYNC_PEERS { + return Err(FileError::Sync(SyncError::PeerLimit)); + } + if peer_id == self.actor_id { + return Err(FileError::Sync(SyncError::InvalidPeer( + "a document cannot sync with its own actor".into(), + ))); + } + let peer = PeerSync::new(peer_id.clone())?; + let status = peer.status(); + let mut peers = self.sync_peers.clone(); + peers.insert(peer_id, peer); + self.persist_sync_peers(&peers)?; + self.sync_peers = peers; + Ok(status) + } + + /// Removes a trusted peer checkpoint. + /// + /// # Errors + /// + /// Returns a typed peer or filesystem error when the checkpoint cannot be + /// safely persisted. + pub fn disconnect_peer(&mut self, peer_id: &ActorId) -> Result<(), FileError> { + let mut peers = self.sync_peers.clone(); + peers.remove(peer_id); + self.persist_sync_peers(&peers)?; + self.sync_peers = peers; + Ok(()) + } + + /// Produces the next message for a connected peer. + /// + /// # Errors + /// + /// Returns a typed peer, CRDT, or filesystem error. + pub fn next_sync_message(&mut self, peer_id: &ActorId) -> Result, FileError> { + let mut peers = self.sync_peers.clone(); + let peer = peers + .get_mut(peer_id) + .ok_or_else(|| FileError::Sync(SyncError::PeerNotConnected { peer_id: peer_id.clone() }))?; + let message = self.engine.next_sync_message(peer)?; + self.persist_sync_peers(&peers)?; + self.sync_peers = peers; + Ok(message) + } + + /// Receives one transport message on a connected peer checkpoint. + /// + /// The transaction engine validates and repairs a fork before this file + /// session adopts it. Peer state is persisted in the same operation, so a + /// restart does not resend an unbounded history. + /// + /// # Errors + /// + /// Returns a typed peer, merge, or filesystem error. + pub fn receive_sync_message(&mut self, message: &SyncMessage) -> Result { + let previous_engine = self.engine.clone(); + let previous_peers = self.sync_peers.clone(); + let before_heads = self.engine.snapshot()?.heads; + let mut peers = self.sync_peers.clone(); + let peer = peers + .get_mut(&message.sender) + .ok_or_else(|| FileError::Sync(SyncError::PeerNotConnected { peer_id: message.sender.clone() }))?; + let mut engine = self.engine.clone(); + let result = engine.receive_sync_message(peer, message)?; + let document_changed = !same_heads(&before_heads, &result.heads); + if document_changed { + self.engine = engine; + self.sync_peers = peers; + if let Err(error) = self.save() { + self.engine = previous_engine; + self.sync_peers = previous_peers; + return Err(error); + } + let peers = self.sync_peers.clone(); + if let Err(error) = self.persist_sync_peers(&peers) { + self.sync_state_warning = Some(format!("could not persist sync checkpoint: {error}")); + } + } else { + self.persist_sync_peers_for_engine(&peers, &mut engine)?; + self.engine = engine; + self.sync_peers = peers; + } + Ok(result) + } + + fn persist_sync_peers(&mut self, peers: &BTreeMap) -> Result<(), FileError> { + let mut engine = self.engine.clone(); + self.persist_sync_peers_for_engine(peers, &mut engine) + } + + fn persist_sync_peers_for_engine( + &self, peers: &BTreeMap, engine: &mut TransactionEngine, + ) -> Result<(), FileError> { + let document_id = engine.snapshot()?.document_id; + let path = sync_state_path_for(&self.path, &document_id, &self.options); + if peers.is_empty() { + match fs::remove_file(&path) { + Ok(()) => Ok(()), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(io_error("remove sync state", path, error)), + } + } else { + let parent = path + .parent() + .ok_or_else(|| FileError::InvalidRecovery("sync state path has no parent".into()))?; + fs::create_dir_all(parent) + .map_err(|error| io_error("create sync state directory", parent.to_owned(), error))?; + let state = SyncStateFile { + format: SYNC_STATE_FORMAT.into(), + version: SYNC_STATE_VERSION, + source_path: path_string(&self.path), + document_id, + actor_id: self.actor_id.clone(), + peers: peers.values().map(PeerSync::persisted).collect(), + }; + write_sync_state(&path, &state) + } + } + + fn load_sync_state(&mut self) { + let path = match self.engine.snapshot() { + Ok(snapshot) => sync_state_path_for(&self.path, &snapshot.document_id, &self.options), + Err(error) => { + self.sync_state_warning = Some(format!("could not inspect sync state: {error}")); + return; + } + }; + let bytes = match fs::read(&path) { + Ok(bytes) => bytes, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return, + Err(error) => { + self.sync_state_warning = Some(format!("could not read {}: {error}", path.display())); + return; + } + }; + if bytes.len() > MAX_SYNC_STATE_BYTES { + self.sync_state_warning = Some(format!("sync state {} exceeds its size limit", path.display())); + return; + } + let state = match serde_json::from_slice::(&bytes) { + Ok(state) => state, + Err(error) => { + self.sync_state_warning = Some(format!("invalid sync state {}: {error}", path.display())); + return; + } + }; + let document_id = match self.engine.snapshot() { + Ok(snapshot) => snapshot.document_id, + Err(error) => { + self.sync_state_warning = Some(format!("could not inspect sync state identity: {error}")); + return; + } + }; + if state.format != SYNC_STATE_FORMAT + || state.version != SYNC_STATE_VERSION + || state.source_path != path_string(&self.path) + || state.document_id != document_id + || state.actor_id != self.actor_id + || state.peers.len() > MAX_SYNC_PEERS + { + self.sync_state_warning = Some(format!("sync state {} does not match this replica", path.display())); + return; + } + for persisted in state.peers { + if let Err(error) = persisted.validate_for(&document_id, &self.actor_id) { + self.sync_state_warning = Some(format!("invalid peer checkpoint envelope: {error}")); + continue; + } + match PeerSync::from_persisted(persisted) { + Ok(peer) if peer.peer_id() != &self.actor_id => { + let peer_id = peer.peer_id().clone(); + match self.sync_peers.entry(peer_id) { + Entry::Occupied(entry) => { + self.sync_state_warning = + Some(format!("sync state contains duplicate peer {}", entry.key())); + } + Entry::Vacant(entry) => { + entry.insert(peer); + } + } + } + Ok(_) => { + self.sync_state_warning = Some("sync state contained the local actor as a peer".into()); + } + Err(error) => { + self.sync_state_warning = Some(format!("invalid peer checkpoint: {error}")); + } + } + } + } + /// Commits one validated transaction through the held document session. /// /// # Errors /// /// Returns [`FileError`] when the transaction engine rejects the draft. - pub fn commit(&mut self, transaction: TransactionDraft) -> Result { + pub fn commit(&mut self, transaction: TransactionDraft) -> Result { Ok(self.engine.commit(transaction)?) } @@ -404,7 +654,7 @@ impl DocumentFile { pub fn save(&mut self) -> Result { let snapshot = self.engine.snapshot()?; let document_id = snapshot.document_id.clone(); - let heads = snapshot.heads.clone(); + let heads = snapshot.heads; let bytes = self.engine.save()?; let recovery_path = recovery_path_for(&self.path, &document_id, &self.options); @@ -479,13 +729,14 @@ impl DocumentFile { return Err(FileError::SamePath { path: self.path.clone() }); } + let previous_path = self.path.clone(); let replacement_lock = AdvisoryLock::acquire(&destination)?; let snapshot = self.engine.snapshot()?; let document_id = snapshot.document_id.clone(); - let heads = snapshot.heads.clone(); + let heads = snapshot.heads; let bytes = self.engine.save()?; let recovery_path = recovery_path_for(&destination, &document_id, &self.options); - let recovery = RecoveryFile::new(&destination, document_id, bytes.clone(), heads.clone()); + let recovery = RecoveryFile::new(&destination, document_id.clone(), bytes.clone(), heads.clone()); let recovery_directory = recovery_path .parent() .ok_or_else(|| FileError::InvalidRecovery("recovery path has no parent".into()))?; @@ -506,6 +757,24 @@ impl DocumentFile { self.pending_recovery = None; self.lock = replacement_lock; + let previous_sync_path = sync_state_path_for(&previous_path, &document_id, &self.options); + let current_sync_path = sync_state_path_for(&self.path, &document_id, &self.options); + let peers = self.sync_peers.clone(); + if let Err(error) = self.persist_sync_peers(&peers) { + self.sync_state_warning = Some(format!("could not persist sync checkpoint after save-as: {error}")); + } else if previous_sync_path != current_sync_path { + match fs::remove_file(&previous_sync_path) { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => { + self.sync_state_warning = Some(format!( + "could not remove old sync checkpoint {}: {error}", + previous_sync_path.display() + )); + } + } + } + Ok(SaveResult { path: self.path.clone(), heads, @@ -615,15 +884,44 @@ pub fn recovery_path_for( .join(format!("{}.recovery", encode_path_component(document_id.as_str()))) } +/// Returns the bounded per-peer synchronization checkpoint path for a +/// document. The source path is part of the filename so two local replicas +/// with the same document ID cannot overwrite one another's checkpoints. +pub fn sync_state_path_for( + document_path: impl AsRef, document_id: &DocumentId, options: &PersistenceOptions, +) -> PathBuf { + let document_path = document_path.as_ref(); + let mut path_hash = 0xcbf2_9ce4_8422_2325_u64; + for byte in path_string(document_path).bytes() { + path_hash ^= u64::from(byte); + path_hash = path_hash.wrapping_mul(0x0100_0000_01b3_u64); + } + options.recovery_directory_for(document_path).join(format!( + "{}.{path_hash:016x}.sync", + encode_path_component(document_id.as_str()) + )) +} + fn create_session_from_engine( path: PathBuf, actor_id: ActorId, mut engine: TransactionEngine, options: PersistenceOptions, ) -> Result { let lock = AdvisoryLock::acquire(&path)?; let baseline_bytes = engine.save()?; let baseline_heads = engine.snapshot()?.heads; - let mut session = - DocumentFile { path, actor_id, engine, options, baseline_bytes, baseline_heads, pending_recovery: None, lock }; + let mut session = DocumentFile { + path, + actor_id, + engine, + options, + baseline_bytes, + baseline_heads, + pending_recovery: None, + sync_peers: BTreeMap::new(), + sync_state_warning: None, + lock, + }; session.save()?; + session.load_sync_state(); Ok(session) } @@ -632,6 +930,16 @@ fn write_recovery(path: &Path, recovery: &RecoveryFile) -> Result<(), FileError> atomic_write(path, &bytes).map(|_| ()) } +fn write_sync_state(path: &Path, state: &SyncStateFile) -> Result<(), FileError> { + let bytes = serde_json::to_vec(state)?; + if bytes.len() > MAX_SYNC_STATE_BYTES { + return Err(FileError::InvalidRecovery(format!( + "sync state exceeds the {MAX_SYNC_STATE_BYTES} byte limit" + ))); + } + atomic_write(path, &bytes).map(|_| ()) +} + fn read_recovery(path: &Path, document_path: &Path, options: &PersistenceOptions) -> Result { let bytes = read_bytes(path, "read recovery record")?; let recovery: RecoveryFile = serde_json::from_slice(&bytes) diff --git a/crates/inkfinite-core/src/ipc/mod.rs b/crates/inkfinite-core/src/ipc/mod.rs index a7959ba..914566c 100644 --- a/crates/inkfinite-core/src/ipc/mod.rs +++ b/crates/inkfinite-core/src/ipc/mod.rs @@ -497,6 +497,7 @@ pub fn session_protocol_error(error: &SessionError) -> ProtocolError { | FileError::RecoveryAhead { .. } | FileError::Engine(_) | FileError::Io { .. } => "document_file_error", + FileError::Sync(_) => "sync_error", }, SessionError::Engine(_) => "document_engine_error", }; diff --git a/crates/inkfinite-core/src/lib.rs b/crates/inkfinite-core/src/lib.rs index 5ba0a24..ab93d2f 100644 --- a/crates/inkfinite-core/src/lib.rs +++ b/crates/inkfinite-core/src/lib.rs @@ -9,6 +9,7 @@ pub mod ipc; pub mod proto; pub mod render; pub mod session; +pub mod sync; use std::collections::BTreeMap; use std::fmt; diff --git a/crates/inkfinite-core/src/session.rs b/crates/inkfinite-core/src/session.rs index 247857b..4464378 100644 --- a/crates/inkfinite-core/src/session.rs +++ b/crates/inkfinite-core/src/session.rs @@ -14,12 +14,13 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use serde::{Deserialize, Serialize}; use thiserror::Error; -use crate::engine::{EngineError, validate_document}; +use crate::engine::{EngineError, SyncApplyResult, validate_document}; use crate::file::{DocumentFile, FileError}; use crate::proto::{ ApplyAuthorization, CommitResult, DocumentPath, Proposal, ProposalId, Query, QueryResult, SaveResult, SessionId, TransactionDraft, TransactionId, Warning, }; +use crate::sync::{PeerSyncStatus, SyncMessage}; use crate::{ActorId, DocumentId, DocumentSnapshot, Origin, Timestamp, blank_document}; /// Maximum number of pending proposals held by one live session. @@ -33,12 +34,19 @@ pub const PROPOSAL_TTL: Duration = Duration::from_secs(5 * 60); /// Monotonic lifetime of a one-time direct-apply authorization. pub const APPLY_AUTHORIZATION_TTL: Duration = Duration::from_secs(2 * 60); -/// State of the session's future synchronization boundary. +/// State of the session's trusted peer synchronization boundary. #[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] #[serde(rename_all = "snake_case", tag = "status")] pub enum SyncState { - /// Peer synchronization is not enabled for this local session yet. + /// No trusted peer checkpoint is configured for this local session. Disabled, + /// Peer checkpoints loaded and retained for this session. + Enabled { + /// Connected peers in stable actor order. + peers: Vec, + /// A checkpoint warning that did not invalidate the document. + warning: Option, + }, } /// Materialized state tracked for one open document session. @@ -63,7 +71,7 @@ pub struct SessionStatus { pub can_undo: bool, /// Whether the session actor can redo its latest compensated transaction. pub can_redo: bool, - /// Synchronization state retained for the later peer layer. + /// Trusted peer synchronization state and bounded checkpoints. pub sync: SyncState, } @@ -94,6 +102,15 @@ pub struct SessionSaved { pub status: SessionStatus, } +/// Result returned after adopting or quarantining a peer message. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct SessionSync { + /// Merge disposition, patch, heads, and repair warnings. + pub sync: SyncApplyResult, + /// Session state after the exchange. + pub status: SessionStatus, +} + /// Recoverable failure from session lookup, authorization, or document I/O. #[derive(Debug, Error)] pub enum SessionError { @@ -192,7 +209,6 @@ pub enum SessionError { struct DocumentSession { file: DocumentFile, - sync: SyncState, proposals: BTreeMap, expired_proposals: BTreeSet, authorizations: BTreeMap, @@ -403,7 +419,7 @@ impl SessionService { let operations = select_operations(&pending.proposal.transaction.operations, operation_positions)?; let transaction = if operation_positions.is_none() { - pending.proposal.transaction.clone() + pending.proposal.transaction } else { let mut transaction = pending.proposal.transaction.clone(); transaction.id = TransactionId(format!("{}:partial", pending.proposal.id.0)); @@ -563,6 +579,61 @@ impl SessionService { Ok(status) } + /// Trusts a peer actor and persists its empty or previously loaded + /// synchronization checkpoint. + /// + /// # Errors + /// + /// Returns a typed session, peer, or filesystem error. + pub fn connect_peer(&mut self, session_id: &SessionId, peer_id: ActorId) -> Result { + let session = self.session_mut(session_id)?; + session.file.connect_peer(peer_id)?; + session.status(session_id) + } + + /// Removes a peer checkpoint without changing the document. + /// + /// # Errors + /// + /// Returns a typed session or filesystem error. + pub fn disconnect_peer( + &mut self, session_id: &SessionId, peer_id: &ActorId, + ) -> Result { + let session = self.session_mut(session_id)?; + session.file.disconnect_peer(peer_id)?; + session.status(session_id) + } + + /// Produces one message for a connected peer. The returned `None` means + /// Automerge is waiting for an acknowledgement or the peer is current. + /// + /// # Errors + /// + /// Returns a typed session, peer, CRDT, or filesystem error. + pub fn next_sync_message( + &mut self, session_id: &SessionId, peer_id: &ActorId, + ) -> Result, SessionError> { + let session = self.session_mut(session_id)?; + Ok(session.file.next_sync_message(peer_id)?) + } + + /// Adopts one peer message through the validated transaction boundary and + /// persists a changed replica before returning to the transport. + /// + /// # Errors + /// + /// Returns a typed session, merge, repair, or filesystem error. Malformed + /// payloads are represented by a quarantined result and do not replace the + /// open valid document. + pub fn receive_sync_message( + &mut self, session_id: &SessionId, message: &SyncMessage, + ) -> Result { + let session = self.session_mut(session_id)?; + let sync = session.file.receive_sync_message(message)?; + let status = session.status(session_id)?; + Ok(SessionSync { sync, status }) + } + /// Closes a session and releases its advisory lock. /// /// Closing deliberately does not save dirty state. Callers can inspect the @@ -603,7 +674,6 @@ impl SessionService { let session_id = SessionId(format!("session:{}", self.next_session_number)); let mut session = DocumentSession { file, - sync: SyncState::Disabled, proposals: BTreeMap::new(), expired_proposals: BTreeSet::new(), authorizations: BTreeMap::new(), @@ -654,7 +724,14 @@ impl DocumentSession { recovery_available, can_undo: self.file.can_undo(), can_redo: self.file.can_redo(), - sync: self.sync.clone(), + sync: { + let sync = self.file.sync_status(); + if sync.peers.is_empty() && sync.warning.is_none() { + SyncState::Disabled + } else { + SyncState::Enabled { peers: sync.peers, warning: sync.warning } + } + }, }) } } @@ -1129,6 +1206,163 @@ mod tests { remove_test_directory(root); } + #[test] + fn two_trusted_replicas_converge_after_offline_edits_and_restart() { + let root = test_directory(); + let left_path = root.join("left.inkfinite"); + let right_path = root.join("right.inkfinite"); + let document_id = DocumentId::from("document:session-sync"); + let left_actor = ActorId::from("actor:left"); + let right_actor = ActorId::from("actor:right"); + let mut left_service = SessionService::new(); + let left = left_service + .create(&left_path, document_id.clone(), left_actor.clone(), None) + .expect("create left"); + fs::copy(&left_path, &right_path).expect("copy baseline to right"); + let mut right_service = SessionService::new(); + let right = right_service + .open(&right_path, right_actor.clone()) + .expect("open right"); + + left_service + .connect_peer(&left.session_id, right_actor.clone()) + .expect("connect left peer"); + right_service + .connect_peer(&right.session_id, left_actor.clone()) + .expect("connect right peer"); + let left_snapshot = left_service.status(&left.session_id).expect("left status").snapshot; + left_service + .commit( + &left.session_id, + rename_transaction(&left_snapshot, left_actor.clone(), "Left offline edit"), + ) + .expect("left offline edit"); + let right_snapshot = right_service.status(&right.session_id).expect("right status").snapshot; + right_service + .commit( + &right.session_id, + rename_transaction(&right_snapshot, right_actor.clone(), "Right offline edit"), + ) + .expect("right offline edit"); + + exchange_sessions( + &mut left_service, + &left.session_id, + &right_actor, + &mut right_service, + &right.session_id, + &left_actor, + ); + assert_eq!( + left_service + .status(&left.session_id) + .expect("left merged status") + .snapshot, + right_service + .status(&right.session_id) + .expect("right merged status") + .snapshot + ); + + left_service.close(&left.session_id).expect("close left"); + right_service.close(&right.session_id).expect("close right"); + + let mut left_restart = SessionService::new(); + let left_reopened = left_restart.open(&left_path, left_actor.clone()).expect("restart left"); + let mut right_restart = SessionService::new(); + let right_reopened = right_restart + .open(&right_path, right_actor.clone()) + .expect("restart right"); + assert!(matches!(left_reopened.status.sync, SyncState::Enabled { .. })); + assert!(matches!(right_reopened.status.sync, SyncState::Enabled { .. })); + + exchange_sessions( + &mut left_restart, + &left_reopened.session_id, + &right_actor, + &mut right_restart, + &right_reopened.session_id, + &left_actor, + ); + assert_eq!( + left_restart + .status(&left_reopened.session_id) + .expect("restarted left status") + .snapshot, + right_restart + .status(&right_reopened.session_id) + .expect("restarted right status") + .snapshot + ); + + left_restart + .close(&left_reopened.session_id) + .expect("close restarted left"); + right_restart + .close(&right_reopened.session_id) + .expect("close restarted right"); + remove_test_directory(root); + } + + #[test] + fn corrupt_sync_checkpoint_warns_without_replacing_valid_document() { + let root = test_directory(); + let path = root.join("corrupt-sync.inkfinite"); + let actor = ActorId::from("actor:session-sync"); + let mut service = SessionService::new(); + let opened = service + .create(&path, DocumentId::from("document:corrupt-sync"), actor.clone(), None) + .expect("create document"); + service + .connect_peer(&opened.session_id, ActorId::from("actor:trusted")) + .expect("connect peer"); + let expected_snapshot = opened.status.snapshot.clone(); + service.close(&opened.session_id).expect("close document"); + + let sync_path = crate::file::sync_state_path_for( + &path, + &expected_snapshot.document_id, + &crate::file::PersistenceOptions::default(), + ); + fs::write(&sync_path, b"not-json").expect("corrupt sync checkpoint"); + + let reopened = service.open(&path, actor).expect("open valid document"); + assert_eq!(reopened.status.snapshot, expected_snapshot); + assert!(matches!( + reopened.status.sync, + SyncState::Enabled { warning: Some(_), .. } + )); + service.close(&reopened.session_id).expect("close reopened document"); + remove_test_directory(root); + } + + fn exchange_sessions( + left: &mut SessionService, left_session: &SessionId, left_peer: &ActorId, right: &mut SessionService, + right_session: &SessionId, right_peer: &ActorId, + ) { + for _ in 0..64 { + let left_message = left + .next_sync_message(left_session, left_peer) + .expect("left sync message"); + let right_message = right + .next_sync_message(right_session, right_peer) + .expect("right sync message"); + let idle = left_message.is_none() && right_message.is_none(); + if let Some(message) = left_message { + right + .receive_sync_message(right_session, &message) + .expect("left to right sync"); + } + if let Some(message) = right_message { + left.receive_sync_message(left_session, &message) + .expect("right to left sync"); + } + if idle { + break; + } + } + } + fn agent_rename(snapshot: &DocumentSnapshot, actor_id: ActorId, name: &str) -> TransactionDraft { let mut transaction = rename_transaction(snapshot, actor_id, name); transaction.origin = Origin::Agent; diff --git a/crates/inkfinite-core/src/sync.rs b/crates/inkfinite-core/src/sync.rs new file mode 100644 index 0000000..673efb1 --- /dev/null +++ b/crates/inkfinite-core/src/sync.rs @@ -0,0 +1,807 @@ +#![forbid(unsafe_code)] + +//! Transport-neutral peer synchronization for Inkfinite documents. +//! +//! Automerge owns the change exchange protocol, while this module owns the +//! Inkfinite envelope, peer identity checks, bounded retry state, and durable +//! checkpoints. A transport only needs to deliver [`SyncMessage`] values and +//! does not need to understand Automerge. + +use std::collections::{BTreeMap, BTreeSet, VecDeque}; + +use serde::{Deserialize, Serialize}; +use thiserror::Error; + +use crate::crdt::{AutomergeDocument, AutomergeSyncSession, CrdtDocument, CrdtError, SyncSession}; +use crate::{ActorId, ChangeHash, DocumentId}; + +/// Stable identifier for the peer synchronization envelope. +pub const SYNC_PROTOCOL_ID: &str = "inkfinite.sync"; +/// Version of the peer synchronization envelope. +pub const SYNC_PROTOCOL_VERSION: u32 = 1; +/// Maximum encoded Automerge sync payload accepted from a peer. +pub const MAX_SYNC_PAYLOAD_BYTES: usize = 8 * 1024 * 1024; +/// Maximum number of messages retained while a transport is delayed or +/// reordered. +pub const MAX_PENDING_SYNC_MESSAGES: usize = 64; +/// Maximum encoded payload bytes retained while a transport is delayed or +/// reordered. +pub const MAX_PENDING_SYNC_BYTES: usize = 8 * 1024 * 1024; +/// Maximum number of peer checkpoints retained for one document. +pub const MAX_SYNC_PEERS: usize = 32; + +const MAX_RECEIVED_SEQUENCES: usize = 4096; + +/// One transport-independent Automerge sync message. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SyncMessage { + /// Stable envelope identifier. + pub protocol_id: String, + /// Envelope version. + pub version: u32, + /// Document this message belongs to. + pub document_id: DocumentId, + /// Actor that produced the message. + pub sender: ActorId, + /// Trusted peer that should receive the message. + pub recipient: ActorId, + /// Monotonic sequence assigned by the sending peer checkpoint. + pub sequence: u64, + /// Opaque Automerge sync payload. + pub payload: Vec, +} + +impl SyncMessage { + fn validate( + &self, document_id: &DocumentId, local_actor: &ActorId, expected_peer: &ActorId, + ) -> Result<(), SyncError> { + if self.protocol_id != SYNC_PROTOCOL_ID || self.version != SYNC_PROTOCOL_VERSION { + return Err(SyncError::UnsupportedProtocol { + protocol_id: self.protocol_id.clone(), + version: self.version, + }); + } + if self.document_id != *document_id { + return Err(SyncError::DocumentMismatch { + expected: document_id.clone(), + actual: self.document_id.clone(), + }); + } + if self.sender != *expected_peer { + return Err(SyncError::PeerMismatch { expected: expected_peer.clone(), actual: self.sender.clone() }); + } + if self.recipient != *local_actor { + return Err(SyncError::WrongRecipient { expected: local_actor.clone(), actual: self.recipient.clone() }); + } + if self.sequence == 0 { + return Err(SyncError::InvalidSequence); + } + if self.payload.is_empty() { + return Err(SyncError::EmptyPayload); + } + if self.payload.len() > MAX_SYNC_PAYLOAD_BYTES { + return Err(SyncError::PayloadTooLarge { size: self.payload.len() }); + } + Ok(()) + } +} + +/// Public state of a trusted peer checkpoint. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct PeerSyncStatus { + /// Trusted actor identity for the peer. + pub peer_id: ActorId, + /// Messages waiting for a delayed dependency or transport delivery. + pub pending_messages: usize, + /// Causal heads the Automerge protocol knows are shared. + pub shared_heads: Vec, + /// Most recently quarantined message, if any. + pub quarantine: Option, +} + +/// A malformed or invalid peer message retained for diagnostics. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct SyncQuarantine { + /// Sequence of the rejected message. + pub sequence: u64, + /// Stable human-readable reason for the rejection. + pub reason: String, +} + +/// Result classification for one received peer message. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum SyncDisposition { + /// At least one new Automerge message was adopted. + Applied, + /// The message sequence was already processed or is already queued. + Duplicate, + /// The message is valid but is waiting for another delayed message. + Deferred, + /// The payload was rejected and retained as quarantine metadata. + Quarantined, +} + +/// Result of receiving one peer envelope before document-level validation. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct PeerReceiveResult { + /// Classification of the received message. + pub disposition: SyncDisposition, + /// Number of queued messages successfully applied while draining pending + /// dependencies. + pub applied_messages: usize, +} + +/// Recoverable failure at the trusted peer boundary. +#[derive(Debug, Error)] +pub enum SyncError { + /// The Automerge adapter rejected or could not decode a payload. + #[error(transparent)] + Crdt(#[from] CrdtError), + /// The envelope protocol is not supported by this peer. + #[error("unsupported sync protocol {protocol_id:?} version {version}")] + UnsupportedProtocol { protocol_id: String, version: u32 }, + /// The message belongs to another document. + #[error("sync message targets document {actual}, expected {expected}")] + DocumentMismatch { expected: DocumentId, actual: DocumentId }, + /// The message came from another actor than the connected peer. + #[error("sync message came from peer {actual}, expected {expected}")] + PeerMismatch { expected: ActorId, actual: ActorId }, + /// The message is addressed to another local actor. + #[error("sync message targets actor {actual}, expected {expected}")] + WrongRecipient { expected: ActorId, actual: ActorId }, + /// Sequence zero is reserved for an uninitialized envelope. + #[error("sync message sequence must be greater than zero")] + InvalidSequence, + /// A sync envelope must carry an encoded Automerge message. + #[error("sync message payload is empty")] + EmptyPayload, + /// An encoded Automerge message exceeded the peer boundary. + #[error("sync payload is {size} bytes; the limit is {MAX_SYNC_PAYLOAD_BYTES}")] + PayloadTooLarge { size: usize }, + /// A peer ID cannot be empty or equal to the local actor. + #[error("sync peer ID is invalid: {0}")] + InvalidPeer(String), + /// The document has reached its bounded peer checkpoint limit. + #[error("the document already has {MAX_SYNC_PEERS} sync peers")] + PeerLimit, + /// A message operation referenced a peer that is not connected. + #[error("sync peer {peer_id} is not connected")] + PeerNotConnected { peer_id: ActorId }, + /// The peer has too many delayed messages queued. + #[error("sync peer pending-message limit exceeded")] + PendingLimit, + /// A persisted peer checkpoint could not be decoded safely. + #[error("invalid persisted sync state: {0}")] + InvalidState(String), +} + +/// A serializable per-peer checkpoint used by the file boundary. +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct PersistedPeerSync { + /// Trusted actor identity for the peer. + pub peer_id: ActorId, + /// Compact Automerge synchronization checkpoint. + pub state: Vec, + /// Sequence to assign to the next outgoing message. + pub next_sequence: u64, + /// Bounded sequence window used for duplicate suppression. + pub received_sequences: Vec, + /// Bounded messages waiting for delayed dependencies. + pub pending_messages: Vec, + /// Most recently quarantined payload. + pub quarantine: Option, +} + +/// Current synchronization state for a document. +#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] +pub struct SyncStatus { + /// Connected trusted peers in stable actor order. + pub peers: Vec, + /// A persisted checkpoint warning that did not invalidate the document. + pub warning: Option, +} + +/// State for one trusted peer and one local document replica. +#[derive(Clone)] +pub struct PeerSync { + peer_id: ActorId, + session: AutomergeSyncSession, + next_sequence: u64, + received_sequences: BTreeSet, + received_order: VecDeque, + pending_messages: BTreeMap, + pending_bytes: usize, + quarantine: Option, +} + +impl PeerSync { + /// Creates a fresh checkpoint for a trusted peer. + /// + /// # Errors + /// + /// Returns [`SyncError::InvalidPeer`] for an empty actor ID. + pub fn new(peer_id: ActorId) -> Result { + validate_peer_id(&peer_id)?; + Ok(Self { + peer_id, + session: AutomergeSyncSession::new(), + next_sequence: 1, + received_sequences: BTreeSet::new(), + received_order: VecDeque::new(), + pending_messages: BTreeMap::new(), + pending_bytes: 0, + quarantine: None, + }) + } + + /// Restores a bounded checkpoint after a process restart. + /// + /// # Errors + /// + /// Returns [`SyncError::InvalidState`] when the encoded checkpoint or its + /// bounded queues cannot be trusted. + pub fn from_persisted(value: PersistedPeerSync) -> Result { + validate_peer_id(&value.peer_id)?; + if value.next_sequence == 0 { + return Err(SyncError::InvalidState( + "next sequence must be greater than zero".into(), + )); + } + if value.received_sequences.len() > MAX_RECEIVED_SEQUENCES { + return Err(SyncError::InvalidState("received sequence window is too large".into())); + } + if value + .quarantine + .as_ref() + .is_some_and(|quarantine| quarantine.sequence == 0) + { + return Err(SyncError::InvalidState( + "quarantine sequence must be greater than zero".into(), + )); + } + let session = AutomergeSyncSession::from_encoded_state(&value.state) + .map_err(|error| SyncError::InvalidState(error.to_string()))?; + let mut peer = Self { + peer_id: value.peer_id, + session, + next_sequence: value.next_sequence, + received_sequences: BTreeSet::new(), + received_order: VecDeque::new(), + pending_messages: BTreeMap::new(), + pending_bytes: 0, + quarantine: value.quarantine, + }; + for sequence in value.received_sequences { + if sequence == 0 || !peer.received_sequences.insert(sequence) { + return Err(SyncError::InvalidState( + "received sequence window contains a duplicate or zero".into(), + )); + } + peer.received_order.push_back(sequence); + } + for message in value.pending_messages { + message.validate_pending(&peer.peer_id)?; + if peer.received_sequences.contains(&message.sequence) { + return Err(SyncError::InvalidState( + "pending message sequence is already marked received".into(), + )); + } + if peer + .pending_messages + .insert(message.sequence, message.clone()) + .is_some() + { + return Err(SyncError::InvalidState( + "pending messages contain a duplicate sequence".into(), + )); + } + peer.pending_bytes = peer + .pending_bytes + .checked_add(message.payload.len()) + .ok_or_else(|| SyncError::InvalidState("pending payload size overflow".into()))?; + if peer.pending_messages.len() > MAX_PENDING_SYNC_MESSAGES || peer.pending_bytes > MAX_PENDING_SYNC_BYTES { + return Err(SyncError::InvalidState( + "pending message queue exceeds its bound".into(), + )); + } + } + Ok(peer) + } + + /// Serializes the bounded checkpoint for persistence. + #[must_use] + pub fn persisted(&self) -> PersistedPeerSync { + PersistedPeerSync { + peer_id: self.peer_id.clone(), + state: self.session.encoded_state(), + next_sequence: self.next_sequence, + received_sequences: self.received_order.iter().copied().collect(), + pending_messages: self.pending_messages.values().cloned().collect(), + quarantine: self.quarantine.clone(), + } + } + + /// Returns the trusted peer actor ID. + #[must_use] + pub fn peer_id(&self) -> &ActorId { + &self.peer_id + } + + /// Returns a stable status projection without exposing Automerge types. + #[must_use] + pub fn status(&self) -> PeerSyncStatus { + PeerSyncStatus { + peer_id: self.peer_id.clone(), + pending_messages: self.pending_messages.len(), + shared_heads: self.session.shared_heads(), + quarantine: self.quarantine.clone(), + } + } + + /// Produces the next outgoing message for this peer. + /// + /// Automerge suppresses messages until its previous in-flight exchange is + /// acknowledged, so callers can safely poll this method after reconnects. + /// + /// # Errors + /// + /// Returns a typed CRDT or sequence-boundary error. + pub fn next_message(&mut self, document: &mut AutomergeDocument) -> Result, SyncError> { + let Some(payload) = self.session.generate_message(document)? else { + return Ok(None); + }; + if payload.len() > MAX_SYNC_PAYLOAD_BYTES { + return Err(SyncError::PayloadTooLarge { size: payload.len() }); + } + let sequence = self.next_sequence; + self.next_sequence = self.next_sequence.checked_add(1).ok_or(SyncError::InvalidSequence)?; + Ok(Some(SyncMessage { + protocol_id: SYNC_PROTOCOL_ID.into(), + version: SYNC_PROTOCOL_VERSION, + document_id: document_id(document)?, + sender: document.actor_id(), + recipient: self.peer_id.clone(), + sequence, + payload, + })) + } + + /// Receives one message, retrying queued messages whose dependencies have + /// arrived and quarantining malformed payloads without mutating the live + /// document on its own. + /// + /// The transaction engine calls this on a document fork, validates and + /// repairs that fork, and adopts both states together. + /// + /// # Errors + /// + /// Returns an error for an envelope addressed to another document, actor, + /// or peer, or when the bounded pending queue is full. + pub fn receive_message( + &mut self, document: &mut AutomergeDocument, message: &SyncMessage, + ) -> Result { + let document_id = document_id(document)?; + message.validate(&document_id, &document.actor_id(), &self.peer_id)?; + if self.received_sequences.contains(&message.sequence) || self.pending_messages.contains_key(&message.sequence) + { + return Ok(PeerReceiveResult { disposition: SyncDisposition::Duplicate, applied_messages: 0 }); + } + if self.pending_messages.len() >= MAX_PENDING_SYNC_MESSAGES + || self.pending_bytes.saturating_add(message.payload.len()) > MAX_PENDING_SYNC_BYTES + { + return Err(SyncError::PendingLimit); + } + self.pending_bytes += message.payload.len(); + self.pending_messages.insert(message.sequence, message.clone()); + + let mut applied_messages = 0; + let mut quarantined = false; + loop { + let pending = self.pending_messages.values().cloned().collect::>(); + let mut made_progress = false; + for pending_message in pending { + let mut candidate_document = document.clone(); + let mut candidate_session = self.session.clone(); + match candidate_session.receive_message_with_status(&mut candidate_document, &pending_message.payload) { + Ok(true) => { + *document = candidate_document; + self.session = candidate_session; + self.remove_pending(pending_message.sequence); + self.record_received(pending_message.sequence); + applied_messages += 1; + made_progress = true; + } + Ok(false) => {} + Err(error) if missing_dependency(&error) => {} + Err(error) => { + self.remove_pending(pending_message.sequence); + self.record_received(pending_message.sequence); + self.quarantine = + Some(SyncQuarantine { sequence: pending_message.sequence, reason: error.to_string() }); + quarantined = true; + made_progress = true; + } + } + } + if !made_progress { + break; + } + } + + let disposition = if quarantined { + SyncDisposition::Quarantined + } else if applied_messages > 0 { + SyncDisposition::Applied + } else { + SyncDisposition::Deferred + }; + Ok(PeerReceiveResult { disposition, applied_messages }) + } + + fn remove_pending(&mut self, sequence: u64) { + if let Some(message) = self.pending_messages.remove(&sequence) { + self.pending_bytes = self.pending_bytes.saturating_sub(message.payload.len()); + } + } + + fn record_received(&mut self, sequence: u64) { + if self.received_sequences.insert(sequence) { + self.received_order.push_back(sequence); + } + while self.received_order.len() > MAX_RECEIVED_SEQUENCES { + if let Some(expired) = self.received_order.pop_front() { + self.received_sequences.remove(&expired); + } + } + } +} + +impl SyncMessage { + fn validate_pending(&self, peer_id: &ActorId) -> Result<(), SyncError> { + if self.protocol_id != SYNC_PROTOCOL_ID || self.version != SYNC_PROTOCOL_VERSION { + return Err(SyncError::InvalidState( + "pending message has an unsupported protocol".into(), + )); + } + if self.sender != *peer_id || self.recipient.as_str().trim().is_empty() || self.sequence == 0 { + return Err(SyncError::InvalidState("pending message identity is invalid".into())); + } + if self.payload.is_empty() || self.payload.len() > MAX_SYNC_PAYLOAD_BYTES { + return Err(SyncError::InvalidState( + "pending message payload is outside its bounds".into(), + )); + } + Ok(()) + } +} + +impl PersistedPeerSync { + /// Validates persisted pending envelopes against their owning document and + /// local actor before the checkpoint is restored. + pub(crate) fn validate_for(&self, document_id: &DocumentId, local_actor: &ActorId) -> Result<(), SyncError> { + for message in &self.pending_messages { + message.validate(document_id, local_actor, &self.peer_id)?; + } + Ok(()) + } +} + +fn validate_peer_id(peer_id: &ActorId) -> Result<(), SyncError> { + if peer_id.as_str().trim().is_empty() { + Err(SyncError::InvalidPeer("actor ID must not be empty".into())) + } else { + Ok(()) + } +} + +fn document_id(document: &mut AutomergeDocument) -> Result { + // The sync envelope needs the stable ID but the CRDT adapter intentionally + // exposes only the materialized snapshot at this boundary. + Ok(document.snapshot()?.document_id) +} + +fn missing_dependency(error: &CrdtError) -> bool { + matches!( + error, + CrdtError::Automerge(source) if matches!(source.as_ref(), automerge::AutomergeError::MissingDeps) + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::crdt::CrdtDocument; + use crate::engine::{TransactionDraft, TransactionEngine}; + use crate::proto::{Operation, TransactionId}; + use crate::{Origin, RecordVersion, Timestamp, blank_document}; + use automerge::transaction::Transactable; + use proptest::prelude::*; + + fn rename(engine: &mut TransactionEngine, actor_id: ActorId, name: &str) { + let snapshot = engine.snapshot().expect("snapshot"); + engine + .commit(TransactionDraft { + id: TransactionId(format!("transaction:{name}")), + actor_id, + origin: Origin::Human, + base_heads: snapshot.heads, + description: name.into(), + operations: vec![Operation::RenamePage { + page_id: snapshot.document.page_ids[0].clone(), + name: name.into(), + expected_version: Some(RecordVersion(1)), + }], + timestamp: Timestamp(1), + }) + .expect("rename"); + } + + fn exchange_until_idle( + left: &mut TransactionEngine, left_peer: &mut PeerSync, right: &mut TransactionEngine, + right_peer: &mut PeerSync, + ) { + for _ in 0..64 { + let left_message = left.next_sync_message(left_peer).expect("left message"); + let right_message = right.next_sync_message(right_peer).expect("right message"); + let idle = left_message.is_none() && right_message.is_none(); + if let Some(message) = left_message { + right.receive_sync_message(right_peer, &message).expect("left to right"); + } + if let Some(message) = right_message { + left.receive_sync_message(left_peer, &message).expect("right to left"); + } + if idle { + break; + } + } + } + + #[test] + fn peer_envelope_rejects_wrong_document_and_recipient() { + let document_id = DocumentId::from("document:sync"); + let mut document = AutomergeDocument::create( + document_id.clone(), + ActorId::from("actor:left"), + blank_document(&document_id, None), + ) + .expect("document"); + let mut peer = PeerSync::new(ActorId::from("actor:right")).expect("peer"); + let mut right_document = AutomergeDocument::create( + document_id.clone(), + ActorId::from("actor:right"), + blank_document(&document_id, None), + ) + .expect("right document"); + let mut right_peer = PeerSync::new(ActorId::from("actor:left")).expect("right peer"); + let message = right_peer + .next_message(&mut right_document) + .expect("message generation") + .expect("handshake"); + let mut wrong = message.clone(); + wrong.document_id = DocumentId::from("document:other"); + assert!(matches!( + peer.receive_message(&mut document, &wrong), + Err(SyncError::DocumentMismatch { .. }) + )); + let mut wrong_recipient = message; + wrong_recipient.recipient = ActorId::from("actor:other"); + assert!(matches!( + peer.receive_message(&mut document, &wrong_recipient), + Err(SyncError::WrongRecipient { .. }) + )); + } + + #[test] + fn persisted_peer_state_is_bounded_and_round_trips() { + let peer = PeerSync::new(ActorId::from("actor:right")).expect("peer"); + let persisted = peer.persisted(); + let restored = PeerSync::from_persisted(persisted).expect("restore"); + assert_eq!(restored.status(), peer.status()); + } + + #[test] + fn two_engines_exchange_only_protocol_messages_and_converge() { + let document_id = DocumentId::from("document:sync"); + let mut left = TransactionEngine::create( + document_id.clone(), + ActorId::from("actor:left"), + blank_document(&document_id, None), + ) + .expect("left"); + let mut right = + TransactionEngine::load(&left.save().expect("save"), ActorId::from("actor:right")).expect("right"); + rename(&mut left, ActorId::from("actor:left"), "left"); + rename(&mut right, ActorId::from("actor:right"), "right"); + let mut left_peer = PeerSync::new(ActorId::from("actor:right")).expect("left peer"); + let mut right_peer = PeerSync::new(ActorId::from("actor:left")).expect("right peer"); + + exchange_until_idle(&mut left, &mut left_peer, &mut right, &mut right_peer); + assert_eq!( + left.snapshot().expect("left snapshot"), + right.snapshot().expect("right snapshot") + ); + assert_eq!(left_peer.status().pending_messages, 0); + assert_eq!(right_peer.status().pending_messages, 0); + } + + #[test] + fn corrupt_payload_is_quarantined_and_duplicate_delivery_is_harmless() { + let document_id = DocumentId::from("document:sync-corrupt"); + let mut left = TransactionEngine::create( + document_id.clone(), + ActorId::from("actor:left"), + blank_document(&document_id, None), + ) + .expect("left"); + let mut right = + TransactionEngine::load(&left.save().expect("save"), ActorId::from("actor:right")).expect("right"); + rename(&mut right, ActorId::from("actor:right"), "right edit"); + let mut left_peer = PeerSync::new(ActorId::from("actor:right")).expect("left peer"); + let mut right_peer = PeerSync::new(ActorId::from("actor:left")).expect("right peer"); + let message = right + .next_sync_message(&mut right_peer) + .expect("message") + .expect("message"); + let before = left.snapshot().expect("before"); + let mut corrupt = message.clone(); + corrupt.payload[0] ^= 0xff; + + let result = left + .receive_sync_message(&mut left_peer, &corrupt) + .expect("quarantine is recoverable"); + assert_eq!(result.disposition, SyncDisposition::Quarantined); + assert_eq!(result.adopted_messages, 0); + assert_eq!(left.snapshot().expect("after"), before); + assert_eq!(left_peer.status().quarantine.unwrap().sequence, message.sequence); + + let duplicate = left + .receive_sync_message(&mut left_peer, &corrupt) + .expect("duplicate quarantine delivery"); + assert_eq!(duplicate.disposition, SyncDisposition::Duplicate); + assert_eq!(left.snapshot().expect("after duplicate"), before); + } + + #[test] + fn reordered_and_duplicated_messages_still_converge() { + let document_id = DocumentId::from("document:sync-reordered"); + let mut left = TransactionEngine::create( + document_id.clone(), + ActorId::from("actor:left"), + blank_document(&document_id, None), + ) + .expect("left"); + let mut right = + TransactionEngine::load(&left.save().expect("save"), ActorId::from("actor:right")).expect("right"); + rename(&mut left, ActorId::from("actor:left"), "left edit"); + rename(&mut right, ActorId::from("actor:right"), "right edit"); + let mut left_peer = PeerSync::new(ActorId::from("actor:right")).expect("left peer"); + let mut right_peer = PeerSync::new(ActorId::from("actor:left")).expect("right peer"); + let left_message = left + .next_sync_message(&mut left_peer) + .expect("left message") + .expect("left message"); + let right_message = right + .next_sync_message(&mut right_peer) + .expect("right message") + .expect("right message"); + + let first = left + .receive_sync_message(&mut left_peer, &right_message) + .expect("right message first"); + assert_eq!(first.disposition, SyncDisposition::Applied); + let duplicate = left + .receive_sync_message(&mut left_peer, &right_message) + .expect("duplicate right message"); + assert_eq!(duplicate.disposition, SyncDisposition::Duplicate); + let first = right + .receive_sync_message(&mut right_peer, &left_message) + .expect("left message second"); + assert_eq!(first.disposition, SyncDisposition::Applied); + let duplicate = right + .receive_sync_message(&mut right_peer, &left_message) + .expect("duplicate left message"); + assert_eq!(duplicate.disposition, SyncDisposition::Duplicate); + + exchange_until_idle(&mut left, &mut left_peer, &mut right, &mut right_peer); + assert_eq!( + left.snapshot().expect("left snapshot"), + right.snapshot().expect("right snapshot") + ); + } + + #[test] + fn delayed_dependency_is_queued_until_its_parent_arrives() { + let document_id = DocumentId::from("document:sync-delayed"); + let mut source = automerge::AutoCommit::new(); + source.put(automerge::ROOT, "value", "parent").expect("parent change"); + source.commit(); + source.put(automerge::ROOT, "value", "child").expect("child change"); + source.commit(); + let changes = source.get_changes(&[]); + assert_eq!(changes.len(), 2); + let parent = changes[0].clone(); + let child = changes[1].clone(); + assert!(child.deps().contains(&parent.hash())); + let parent_payload = automerge::sync::Message { + heads: vec![parent.hash()], + need: Vec::new(), + have: Vec::new(), + changes: vec![parent.raw_bytes().to_vec()].into(), + flags: None, + version: automerge::sync::MessageVersion::V1, + } + .encode(); + let child_payload = automerge::sync::Message { + heads: vec![child.hash()], + need: vec![parent.hash()], + have: Vec::new(), + changes: vec![child.raw_bytes().to_vec()].into(), + flags: None, + version: automerge::sync::MessageVersion::V1, + } + .encode(); + + let mut document = AutomergeDocument::create( + document_id.clone(), + ActorId::from("actor:left"), + blank_document(&document_id, None), + ) + .expect("document"); + let mut peer = PeerSync::new(ActorId::from("actor:right")).expect("peer"); + let delayed = SyncMessage { + protocol_id: SYNC_PROTOCOL_ID.into(), + version: SYNC_PROTOCOL_VERSION, + document_id: document_id.clone(), + sender: ActorId::from("actor:right"), + recipient: ActorId::from("actor:left"), + sequence: 2, + payload: child_payload, + }; + let parent = SyncMessage { sequence: 1, payload: parent_payload, ..delayed.clone() }; + + let deferred = peer.receive_message(&mut document, &delayed).expect("defer child"); + assert_eq!(deferred.disposition, SyncDisposition::Deferred); + assert_eq!(deferred.applied_messages, 0); + assert_eq!(peer.status().pending_messages, 1); + let applied = peer + .receive_message(&mut document, &parent) + .expect("apply parent and child"); + assert_eq!(applied.disposition, SyncDisposition::Applied); + assert_eq!(applied.applied_messages, 2); + assert_eq!(peer.status().pending_messages, 0); + } + + proptest! { + #![proptest_config(ProptestConfig::with_cases(8))] + + #[test] + fn concurrent_peer_edits_converge_for_generated_names( + left_name in "[a-z]{1,16}", + right_name in "[a-z]{1,16}", + ) { + let document_id = DocumentId::from("document:sync-property"); + let mut left = TransactionEngine::create( + document_id.clone(), + ActorId::from("actor:left"), + blank_document(&document_id, None), + ).expect("left"); + let mut right = TransactionEngine::load( + &left.save().expect("save"), + ActorId::from("actor:right"), + ).expect("right"); + rename(&mut left, ActorId::from("actor:left"), &left_name); + rename(&mut right, ActorId::from("actor:right"), &right_name); + let mut left_peer = PeerSync::new(ActorId::from("actor:right")).expect("left peer"); + let mut right_peer = PeerSync::new(ActorId::from("actor:left")).expect("right peer"); + + exchange_until_idle(&mut left, &mut left_peer, &mut right, &mut right_peer); + + prop_assert_eq!(left.snapshot().expect("left snapshot"), right.snapshot().expect("right snapshot")); + prop_assert_eq!(left_peer.status().pending_messages, 0); + prop_assert_eq!(right_peer.status().pending_messages, 0); + } + } +}