feat(web): session-safe sync recovery, SSE fan-in, and collab undo for the browser host
Second collaboration increment on the wasm side: - a 409 version conflict during an Active session now auto-resolves by accepting the sequenced daemon document (rejected local edits stay recoverable via the discarded-edit projection); every other phase keeps the explicit-resolve latch, with the decision pinned by tests - the collab tick subscribes to /api/mcp/events via EventSource and feeds the same version/collabSeq latch as the poll, with exponential backoff down to polling when the stream drops - Cmd+Z during a session routes through a synchronous RequestUndo wire command (claimed => session undo, else local); redo raises the M1 Unsupported notice - the session wire carries the peer id namespace and the web host gains a collab id allocator (canvas create wired; remaining creation sites stay behind the PR4 push gate until converted)
This commit is contained in:
Kayshen-X
parent
2bd181bd60
commit
b8e41fa3c6
|
|
@ -139,7 +139,7 @@ Mutators on `Document`:
|
|||
| LayerContextMenu | Right-click overlay on layer rows + page tabs (Rename / Duplicate / Delete / Group / Ungroup / Lock / Hide; subset on page tabs) | `layer_context_menu.rs` |
|
||||
| AgentSettings | Cmd+, modal — 880×640 with sidebar nav (Agents / MCP / Images / System) + scrollable right pane | `agent_settings_panel.rs` + `agent_settings_{i18n,images,mcp,system}.rs` |
|
||||
| theme | shadcn-dark palette tokens (incl. `canvas_surface`) | `theme.rs` |
|
||||
| i18n | 15 canonical locale catalogs, 1,105 direct keys each, sharded main → `_git` → `_panel` in the `op-i18n` crate | `op-i18n/src/i18n/{en,zh_cn,zh_tw,ja,ko,fr,es,de,pt,ru,hi,tr,th,vi,id}{,_git,_panel}.rs` |
|
||||
| i18n | 15 canonical locale catalogs, 1,526 direct keys each, sharded main → `_git` → `_panel` → `_collab` in the `op-i18n` crate | `op-i18n/src/i18n/{en,zh_cn,zh_tw,ja,ko,fr,es,de,pt,ru,hi,tr,th,vi,id}{,_git,_panel,_collab}.rs` |
|
||||
|
||||
## Theme + i18n
|
||||
|
||||
|
|
@ -150,15 +150,15 @@ Mutators on `Document`:
|
|||
- 15 supported locales (matches TS dropdown order): EnUs / ZhCn / ZhTw / Ja / Ko / Fr / Es / De / Pt / Ru / Hi / Tr / Th / Vi / Id. Each carries a `display_name()` (English / 简体中文 / 繁體中文 / 日本語 / 한국어 / Français / Español / Deutsch / Português / Русский / हिन्दी / Türkçe / ไทย / Tiếng Việt / Bahasa Indonesia).
|
||||
- TopBar Globe-button is a 44 px-wide compound (globe + chevron-down) opening a `LocalePicker` dropdown — clicking a row sets `Document.ui.locale` and closes; clicking outside (or the Globe again) closes silently. The picker paints as the top-most overlay so it covers chat / status / canvas.
|
||||
- Multi-script chrome strings (한국어 / हिन्दी / ไทย / Tiếng Việt) render against per-codepoint typeface lookups (`FontMgr::match_family_style_character` cached per `i32` in `NativeBackend`), with each string broken into contiguous-typeface segments before draw.
|
||||
- Locale tables are hand-maintained in Rust, and each locale is **three files, chained by fall-through**: the main table `<locale>.rs` ends its `match` with `_ => return super::<locale>_git::lookup(key)`, the Git shard ends with `_ => return super::<locale>_panel::lookup(key)`, and the `_panel` overflow shard ends with `_ => return None`. The catalog is 1,105 direct keys per locale (`catalog_integrity_tests.rs` asserts the exact number).
|
||||
- Locale tables are hand-maintained in Rust, and each locale is **four files, chained by fall-through**: the main table `<locale>.rs` ends its `match` with `_ => return super::<locale>_git::lookup(key)`, the Git shard ends with `_ => return super::<locale>_panel::lookup(key)`, the `_panel` shard ends with `_ => return super::<locale>_collab::lookup(key)`, and the `_collab` overflow shard ends with `_ => return None`. The catalog is 1,526 direct keys per locale (`catalog_integrity_tests.rs` asserts the exact number).
|
||||
|
||||
**Adding a new key: put it in `<locale>_panel.rs`, in all 15 locales.** The main tables sit at or just under the repo's 800-line ceiling (`zh_cn.rs` is exactly 800, `en.rs` 798), so a new entry there does not fit — the `_panel` shards exist precisely to absorb overflow and have room. Only edit a main or `_git` table when you are changing the wording of a key that already lives there. Then bump the expected count in `catalog_integrity_tests.rs` and run:
|
||||
**Adding a new key: put it in `<locale>_collab.rs` (the terminal overflow shard), in all 15 locales.** The main tables sit at or just under the repo's 800-line ceiling (`zh_cn.rs` is exactly 800, `en.rs` 798), so a new entry there does not fit — the `_panel` shards exist precisely to absorb overflow and have room. Only edit a main or `_git` table when you are changing the wording of a key that already lives there. Then bump the expected count in `catalog_integrity_tests.rs` and run:
|
||||
|
||||
```sh
|
||||
cargo test -p op-i18n
|
||||
```
|
||||
|
||||
Catalog integrity tests enforce the exact cross-locale key set across all three shards, reject duplicate or unsupported match arms, and require every translation to preserve the English placeholder set. The `i18n/mod.rs` tests were split into sibling modules alongside it (`tests.rs`, `catalog_integrity_tests.rs`, plus per-feature key guards: `figma_property_panel_key_tests.rs`, `html_import_key_tests.rs`, `missing_fonts_key_tests.rs`, `preview_device_key_tests.rs`, `vector_fidelity_property_keys.rs`) — add a feature's key guard next to those rather than growing `mod.rs`. `tools/convert-locales.py` is a deliberate failing shim because its retired TypeScript source no longer exists. Runtime lookup still falls back through English and then the raw key for debug visibility, while the tests prevent shipped locale gaps.
|
||||
Catalog integrity tests enforce the exact cross-locale key set across all four shards, reject duplicate or unsupported match arms, and require every translation to preserve the English placeholder set. The `i18n/mod.rs` tests were split into sibling modules alongside it (`tests.rs`, `catalog_integrity_tests.rs`, plus per-feature key guards: `figma_property_panel_key_tests.rs`, `html_import_key_tests.rs`, `missing_fonts_key_tests.rs`, `preview_device_key_tests.rs`, `vector_fidelity_property_keys.rs`) — add a feature's key guard next to those rather than growing `mod.rs`. `tools/convert-locales.py` is a deliberate failing shim because its retired TypeScript source no longer exists. Runtime lookup still falls back through English and then the raw key for debug visibility, while the tests prevent shipped locale gaps.
|
||||
|
||||
## Toolbar shape-tool dropdown
|
||||
|
||||
|
|
|
|||
|
|
@ -254,6 +254,20 @@ impl CollabRuntime {
|
|||
true
|
||||
}
|
||||
|
||||
/// Owner-assigned id namespace for this peer, once a session assigned one.
|
||||
///
|
||||
/// A remote client that creates nodes has to mint their ids from this, or
|
||||
/// two peers minting from private counters eventually agree on an id for
|
||||
/// two different nodes. The daemon publishes it in the collaboration
|
||||
/// projection so the browser can allocate correctly.
|
||||
pub fn peer_namespace(&self) -> Option<String> {
|
||||
let actor = self.actor.as_ref()?;
|
||||
Some(match actor {
|
||||
EditorActor::Owner(owner) => owner.peer_namespace().as_str().to_owned(),
|
||||
EditorActor::Guest(guest) => guest.session.core().peer_namespace().as_str().to_owned(),
|
||||
})
|
||||
}
|
||||
|
||||
pub fn take_save_as_fork_request(&mut self) -> bool {
|
||||
std::mem::take(&mut self.save_as_fork_requested)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -28,6 +28,10 @@ pub(super) struct OwnerActor {
|
|||
pub(super) connections: HashSet<ConnectionKey>,
|
||||
local_connection: ConnectionKey,
|
||||
share_endpoint: Option<CollabShareEndpoint>,
|
||||
/// The owner's own id namespace. The session core tracks each *guest*'s
|
||||
/// namespace; the owner mints its own at start and nothing else records it,
|
||||
/// so it is kept here for the hosts that have to publish it.
|
||||
namespace: PeerNamespace,
|
||||
}
|
||||
|
||||
pub(super) struct GuestActor {
|
||||
|
|
@ -51,6 +55,7 @@ impl OwnerActor {
|
|||
let participant_id = ParticipantId::from(random_identifier("participant")?);
|
||||
let peer_id = PeerId::from(random_identifier("peer")?);
|
||||
let namespace = random_namespace()?;
|
||||
let owner_namespace = namespace.clone();
|
||||
let principal =
|
||||
ConnectionPrincipal::from_verified(auth, participant_id, peer_id, Role::Owner);
|
||||
let owner_connection =
|
||||
|
|
@ -74,9 +79,15 @@ impl OwnerActor {
|
|||
connections: HashSet::new(),
|
||||
local_connection: owner_connection,
|
||||
share_endpoint: None,
|
||||
namespace: owner_namespace,
|
||||
})
|
||||
}
|
||||
|
||||
/// The owner's own id namespace.
|
||||
pub(super) fn peer_namespace(&self) -> &PeerNamespace {
|
||||
&self.namespace
|
||||
}
|
||||
|
||||
pub(super) fn set_share_endpoint(&mut self, endpoint: Option<SocketAddr>) {
|
||||
self.share_endpoint =
|
||||
endpoint.and_then(|endpoint| CollabShareEndpoint::new(endpoint.to_string()));
|
||||
|
|
|
|||
|
|
@ -21,7 +21,7 @@
|
|||
mod action;
|
||||
mod parts;
|
||||
|
||||
pub use action::{CollabActionWire, CollabActionWireError};
|
||||
pub use action::{CollabActionWire, CollabActionWireError, CollabWireCommand};
|
||||
pub use parts::{
|
||||
CollabAdmissionWire, CollabAvailabilityWire, CollabConnectErrorWire, CollabConnectionPathWire,
|
||||
CollabDiscardedEditWire, CollabDiscoveredWire, CollabLocalPresenceWire, CollabNoticeKindWire,
|
||||
|
|
@ -64,6 +64,18 @@ pub struct CollabSessionWire {
|
|||
pub share_endpoint: Option<String>,
|
||||
pub invite: Option<String>,
|
||||
pub connection: Option<CollabConnectionPathWire>,
|
||||
/// Owner-assigned id namespace for this peer.
|
||||
///
|
||||
/// A client that creates nodes MUST mint their ids from this namespace.
|
||||
/// The protocol replays ids verbatim, so two peers minting from a private
|
||||
/// sequential counter would eventually agree on an id for two different
|
||||
/// nodes and silently fork the document.
|
||||
///
|
||||
/// Additive and optional: a client that finds it absent — an older daemon,
|
||||
/// or a session whose namespace is not yet known — must refuse to create
|
||||
/// nodes rather than fall back to a local counter.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub peer_namespace: Option<String>,
|
||||
}
|
||||
|
||||
/// The whole collaboration projection for one poll.
|
||||
|
|
@ -106,6 +118,10 @@ impl CollabStateWire {
|
|||
connection: public
|
||||
.and_then(|public| public.connection())
|
||||
.map(Into::into),
|
||||
// Not part of the shared UI projection — it lives in the
|
||||
// session actor, so the daemon layers it on with
|
||||
// [`Self::with_peer_namespace`].
|
||||
peer_namespace: None,
|
||||
}
|
||||
});
|
||||
Self {
|
||||
|
|
@ -154,6 +170,21 @@ impl CollabStateWire {
|
|||
}
|
||||
}
|
||||
|
||||
/// Attach the owner-assigned id namespace to an already-built projection.
|
||||
///
|
||||
/// Separate from [`Self::from_ui`] because the namespace is owned by the
|
||||
/// session actor rather than the UI state the panel paints; only the host
|
||||
/// running the session can supply it. A projection with no session, or a
|
||||
/// session whose namespace is not yet assigned, keeps `None` — and a client
|
||||
/// reading `None` must not create nodes.
|
||||
#[must_use]
|
||||
pub fn with_peer_namespace(mut self, namespace: Option<String>) -> Self {
|
||||
if let Some(session) = self.session.as_mut() {
|
||||
session.peer_namespace = namespace;
|
||||
}
|
||||
self
|
||||
}
|
||||
|
||||
/// Install this projection into a client's UI state.
|
||||
///
|
||||
/// Every write goes through a sanitising `set_*` / `publish_*` method, and
|
||||
|
|
|
|||
|
|
@ -26,6 +26,22 @@ pub enum CollabActionWireError {
|
|||
InvalidRequestKey,
|
||||
/// A discovery id or endpoint was empty or over [`MAX_WIRE_ADDRESS_CHARS`].
|
||||
InvalidAddress,
|
||||
/// The action is a direct runtime request, not a queued panel action.
|
||||
NotAUiAction,
|
||||
}
|
||||
|
||||
/// What a posted action asks the host to do.
|
||||
///
|
||||
/// Most actions are queued into the panel's pending slot and drained by the
|
||||
/// runtime pump; undo is a direct call on the session. Splitting them keeps
|
||||
/// the caller from having to special-case a variant the pending slot cannot
|
||||
/// represent.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum CollabWireCommand {
|
||||
/// Queue this into `CollabUiState::pending_action`.
|
||||
Ui(CollabUiAction),
|
||||
/// Call the runtime's selective-undo entry point.
|
||||
RequestUndo,
|
||||
}
|
||||
|
||||
impl std::fmt::Display for CollabActionWireError {
|
||||
|
|
@ -33,6 +49,7 @@ impl std::fmt::Display for CollabActionWireError {
|
|||
match self {
|
||||
Self::InvalidRequestKey => f.write_str("malformed collaboration admission key"),
|
||||
Self::InvalidAddress => f.write_str("malformed collaboration address"),
|
||||
Self::NotAUiAction => f.write_str("action is not a queued panel action"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -69,6 +86,12 @@ pub enum CollabActionWire {
|
|||
DiscardPending,
|
||||
ReapplyDiscarded,
|
||||
SaveAsFork,
|
||||
/// Ask the session for an M1 selective undo of this peer's latest edit.
|
||||
///
|
||||
/// Additive: a daemon that does not know this variant rejects the body,
|
||||
/// which is the correct answer — the client must then fall back to a local
|
||||
/// undo rather than assume the request landed.
|
||||
RequestUndo,
|
||||
#[serde(rename_all = "camelCase")]
|
||||
ApproveAdmissionEditor {
|
||||
request_key: String,
|
||||
|
|
@ -109,9 +132,22 @@ impl CollabActionWire {
|
|||
)
|
||||
}
|
||||
|
||||
/// Validate and convert into the internal action.
|
||||
/// Validate and convert into the internal command.
|
||||
pub fn into_command(self) -> Result<CollabWireCommand, CollabActionWireError> {
|
||||
if matches!(self, Self::RequestUndo) {
|
||||
return Ok(CollabWireCommand::RequestUndo);
|
||||
}
|
||||
Ok(CollabWireCommand::Ui(self.into_ui_action()?))
|
||||
}
|
||||
|
||||
/// Validate and convert into the internal UI action.
|
||||
///
|
||||
/// `RequestUndo` has no `CollabUiAction` form — the runtime exposes undo as
|
||||
/// a direct call, not a queued panel action — so it is the one variant this
|
||||
/// cannot produce. Use [`Self::into_command`] to handle both.
|
||||
pub fn into_ui_action(self) -> Result<CollabUiAction, CollabActionWireError> {
|
||||
Ok(match self {
|
||||
Self::RequestUndo => return Err(CollabActionWireError::NotAUiAction),
|
||||
Self::OpenCreate => CollabUiAction::OpenCreate,
|
||||
Self::Start => CollabUiAction::Start,
|
||||
Self::StartLan => CollabUiAction::StartLan,
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@
|
|||
|
||||
use op_editor_core::collab_wire::{
|
||||
CollabActionWire, CollabActionWireError, CollabLocalPresenceWire, CollabStateWire,
|
||||
CollabWireCommand,
|
||||
};
|
||||
|
||||
use super::{WebCanvasState, WebReply};
|
||||
|
|
@ -19,11 +20,16 @@ const MAX_COLLAB_BODY_BYTES: usize = 8 * 1024;
|
|||
|
||||
/// `GET /api/collab/state` — the whole projection plus both sequence numbers.
|
||||
pub(crate) fn state(state: &mut WebCanvasState) -> WebReply {
|
||||
// The namespace lives in the session actor, not the UI projection, so it
|
||||
// is layered on here. A client that creates nodes needs it to mint ids the
|
||||
// protocol can replay.
|
||||
let namespace = state.collab.runtime.peer_namespace();
|
||||
let wire = CollabStateWire::from_ui(
|
||||
&state.editor.editor_ui.collab,
|
||||
state.collab.seq(),
|
||||
state.editor.document_revision(),
|
||||
);
|
||||
)
|
||||
.with_peer_namespace(namespace);
|
||||
match serde_json::to_string(&wire) {
|
||||
Ok(body) => WebReply {
|
||||
status: "200 OK",
|
||||
|
|
@ -60,10 +66,33 @@ pub(crate) fn action(body: &str, state: &mut WebCanvasState) -> WebReply {
|
|||
// a caller-named socket address or enumerate the host's LAN. The local and
|
||||
// managed daemons allow them — that is desktop parity, and the operator is
|
||||
// the only client.
|
||||
let action = match wire.into_ui_action() {
|
||||
Ok(action) => action,
|
||||
let command = match wire.into_command() {
|
||||
Ok(command) => command,
|
||||
Err(error) => return action_error_reply(error),
|
||||
};
|
||||
// Undo is a direct call on the session, not a queued panel action: the
|
||||
// runtime exposes it as a method and the pending slot cannot represent it.
|
||||
// It also runs here rather than on the driver thread because the answer —
|
||||
// whether a session claimed the keystroke — is what the browser needs to
|
||||
// decide between a collaborative undo and a local one.
|
||||
let action = match command {
|
||||
CollabWireCommand::RequestUndo => {
|
||||
let (runtime, mut host) = state.collab_runtime_and_host();
|
||||
let claimed = runtime.request_undo(&mut host);
|
||||
let seq = state.collab.bump_seq();
|
||||
state.collab.wake_driver();
|
||||
return WebReply {
|
||||
status: "200 OK",
|
||||
body: serde_json::json!({
|
||||
"ok": true,
|
||||
"claimed": claimed,
|
||||
"collabSeq": seq,
|
||||
})
|
||||
.to_string(),
|
||||
};
|
||||
}
|
||||
CollabWireCommand::Ui(action) => action,
|
||||
};
|
||||
// The pending slot holds exactly one action, and the runtime consumes it
|
||||
// through `take_pending_action`. Overwriting an undrained action would
|
||||
// silently drop whatever the user asked for first, so this reports the
|
||||
|
|
@ -120,6 +149,7 @@ fn action_error_reply(error: CollabActionWireError) -> WebReply {
|
|||
let code = match error {
|
||||
CollabActionWireError::InvalidRequestKey => "invalid-request-key",
|
||||
CollabActionWireError::InvalidAddress => "invalid-address",
|
||||
CollabActionWireError::NotAUiAction => "unsupported-action",
|
||||
};
|
||||
error_reply("400 Bad Request", code, &error.to_string())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -101,6 +101,11 @@ features = [
|
|||
"Element",
|
||||
"Event",
|
||||
"EventTarget",
|
||||
# `EventSource` — the daemon's SSE channel (`GET /api/mcp/events`). It is a
|
||||
# latency accelerator over the existing version poll, never a replacement:
|
||||
# the poll stays armed so a proxy that buffers SSE, or a dropped stream,
|
||||
# degrades to the old cadence instead of going silent.
|
||||
"EventSource",
|
||||
"File",
|
||||
"FileList",
|
||||
"FileReader",
|
||||
|
|
|
|||
|
|
@ -643,15 +643,22 @@ pub(super) async fn mount_ck(canvas_id: String) -> Result<(), JsValue> {
|
|||
// Case-insensitive: with Shift held, `key` is layout/IME
|
||||
// dependent — macOS Chromium can report either "z" or "Z"
|
||||
// for Cmd+Shift+Z, so branch on the shift flag alone.
|
||||
// In a live session history belongs to the session, not this
|
||||
// tab: undo becomes an M1 selective-undo request the daemon
|
||||
// sequences, and redo is refused outright. Both short-circuit
|
||||
// local history exactly as the desktop host does.
|
||||
"z" | "Z" if is_mod && !image_popover_open => {
|
||||
consumed = if shift {
|
||||
b.host.apply_redo()
|
||||
crate::collab_sync::reject_redo(b.host.editor_state_mut())
|
||||
|| b.host.apply_redo()
|
||||
} else {
|
||||
b.host.apply_undo()
|
||||
crate::collab_sync::request_undo(b.host.editor_state_mut())
|
||||
|| b.host.apply_undo()
|
||||
};
|
||||
}
|
||||
"y" | "Y" if is_mod && !shift && !image_popover_open => {
|
||||
consumed = b.host.apply_redo()
|
||||
consumed = crate::collab_sync::reject_redo(b.host.editor_state_mut())
|
||||
|| b.host.apply_redo()
|
||||
}
|
||||
"s" | "S" if is_mod && !shift => {
|
||||
// VS Code embed: the workbench cannot observe keystrokes
|
||||
|
|
|
|||
|
|
@ -62,6 +62,8 @@ thread_local! {
|
|||
static STATE_BUSY: Cell<bool> = const { Cell::new(false) };
|
||||
/// An action already posted and not yet answered.
|
||||
static ACTION_BUSY: Cell<bool> = const { Cell::new(false) };
|
||||
/// A Cmd+Z is waiting to be posted as a `RequestUndo` action.
|
||||
static UNDO_REQUESTED: Cell<bool> = const { Cell::new(false) };
|
||||
/// An action the daemon refused with `collab-busy`, kept for the next tick.
|
||||
/// Losing it would silently drop something the user clicked.
|
||||
static ACTION_RETRY: RefCell<Option<CollabUiAction>> = const { RefCell::new(None) };
|
||||
|
|
@ -137,6 +139,9 @@ pub(crate) fn start<C: RepaintContext + 'static>(inner: &Rc<RefCell<C>>) {
|
|||
let base = crate::daemon_base::daemon_base();
|
||||
let inner = inner.clone();
|
||||
let tick: Rc<dyn Fn()> = Rc::new(move || {
|
||||
// Re-armed from the tick so a stream that dropped comes back without a
|
||||
// timer of its own; the backoff inside decides whether to actually try.
|
||||
ensure_event_stream(&base);
|
||||
drain_pending_action(&inner, &base);
|
||||
maybe_pull_state(&inner, &base);
|
||||
maybe_push_presence(&inner, &base);
|
||||
|
|
@ -182,6 +187,22 @@ fn drain_pending_action<C: RepaintContext + 'static>(inner: &Rc<RefCell<C>>, bas
|
|||
if ACTION_BUSY.get() {
|
||||
return;
|
||||
}
|
||||
// Undo jumps the queue: it is a keystroke the user just pressed, while the
|
||||
// pending slot holds panel actions that can wait a tick.
|
||||
if UNDO_REQUESTED.replace(false) {
|
||||
ACTION_BUSY.set(true);
|
||||
let body = serde_json::to_string(&CollabActionWire::RequestUndo)
|
||||
.expect("a unit variant always serializes");
|
||||
let started = live_sync::post_json_with_status(
|
||||
&format!("{base}{}", collab_routes::ACTION),
|
||||
&body,
|
||||
Rc::new(move |_status, _response| ACTION_BUSY.set(false)),
|
||||
);
|
||||
if !started {
|
||||
ACTION_BUSY.set(false);
|
||||
}
|
||||
return;
|
||||
}
|
||||
let action = ACTION_RETRY
|
||||
.with(|slot| slot.borrow_mut().take())
|
||||
.or_else(|| {
|
||||
|
|
@ -323,10 +344,45 @@ fn apply_state<C: RepaintContext + 'static>(inner: &Rc<RefCell<C>>, wire: &Colla
|
|||
wire.apply_to(&mut state.editor_ui.collab, now_ms);
|
||||
APPLIED_SEQ.set(Some(wire.collab_seq));
|
||||
SESSION_LIVE.set(state.editor_ui.collab.phase != CollabConnectionPhase::Idle);
|
||||
sync_id_allocation(context.host_mut(), wire);
|
||||
context.host_mut().mark_editor_state_dirty();
|
||||
let _ = context.repaint();
|
||||
}
|
||||
|
||||
/// Follow the projection into and out of namespaced id allocation.
|
||||
///
|
||||
/// Enabled only while a session is `Active` *and* the daemon published a
|
||||
/// namespace; anything else restores the standalone counter. A session whose
|
||||
/// namespace is absent — an older daemon — therefore keeps the local counter,
|
||||
/// and `push_blocked_by_session` is what stops those ids from reaching a peer.
|
||||
fn sync_id_allocation(host: &mut crate::widget_host::WidgetHost, wire: &CollabStateWire) {
|
||||
let namespace = (wire.phase == op_editor_core::collab_wire::CollabPhaseWire::Active)
|
||||
.then(|| wire.session.as_ref().and_then(|s| s.peer_namespace.clone()))
|
||||
.flatten();
|
||||
match namespace {
|
||||
Some(namespace) if !host.collaboration_ids_enabled() => {
|
||||
match op_editor_core::PeerNamespace::parse(namespace) {
|
||||
Ok(namespace) => {
|
||||
if let Err(error) = host.enable_collaboration_ids(namespace) {
|
||||
// The document already carries ids this namespace
|
||||
// cannot resume above. Staying on the standalone
|
||||
// counter keeps the canvas usable; the push gate is
|
||||
// what keeps those ids off the wire.
|
||||
let _ = error;
|
||||
}
|
||||
}
|
||||
Err(_) => host.disable_collaboration_ids(),
|
||||
}
|
||||
}
|
||||
Some(_) => {}
|
||||
None => {
|
||||
if host.collaboration_ids_enabled() {
|
||||
host.disable_collaboration_ids();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn maybe_push_presence<C: RepaintContext + 'static>(inner: &Rc<RefCell<C>>, base: &str) {
|
||||
if !SESSION_LIVE.get() {
|
||||
return;
|
||||
|
|
@ -508,3 +564,127 @@ mod tests {
|
|||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Route Cmd/Ctrl+Z into the session, returning whether the session claimed it.
|
||||
///
|
||||
/// `false` means no session owns the document and the caller should run local
|
||||
/// history — the same short-circuit contract the desktop host uses
|
||||
/// (`collab_runtime.request_undo(host) || host.apply_undo()`).
|
||||
///
|
||||
/// The request is queued as a wire action rather than answered here: undo has
|
||||
/// to be sequenced against the other peers, and only the daemon can do that.
|
||||
pub(crate) fn request_undo(state: &mut op_editor_core::EditorState) -> bool {
|
||||
if state.editor_ui.collab.phase != CollabConnectionPhase::Active {
|
||||
return false;
|
||||
}
|
||||
UNDO_REQUESTED.set(true);
|
||||
true
|
||||
}
|
||||
|
||||
/// Refuse redo while a session is live.
|
||||
///
|
||||
/// M1 collaboration sequences a selective undo per peer but has no matching
|
||||
/// redo, so the honest answer is a notice rather than a local redo that would
|
||||
/// diverge this tab from everyone else.
|
||||
pub(crate) fn reject_redo(state: &mut op_editor_core::EditorState) -> bool {
|
||||
if state.editor_ui.collab.phase != CollabConnectionPhase::Active {
|
||||
return false;
|
||||
}
|
||||
state.editor_ui.collab.set_notice(
|
||||
op_editor_core::CollabNoticeKind::Reject(op_editor_core::CollabRejectUiCode::Unsupported),
|
||||
now_ms(),
|
||||
);
|
||||
true
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// SSE acceleration
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Backoff after a dropped stream, doubling to [`SSE_MAX_RETRY_MS`].
|
||||
const SSE_BASE_RETRY_MS: f64 = 2_000.0;
|
||||
/// Ceiling for the reconnect backoff. Past this the poll is carrying the load
|
||||
/// perfectly well, so retrying harder buys nothing.
|
||||
const SSE_MAX_RETRY_MS: f64 = 60_000.0;
|
||||
|
||||
thread_local! {
|
||||
/// The live stream, when one is open.
|
||||
static EVENT_STREAM: RefCell<Option<web_sys::EventSource>> = const { RefCell::new(None) };
|
||||
/// Earliest `performance.now()` at which a reconnect may be attempted.
|
||||
static SSE_RETRY_AT_MS: Cell<f64> = const { Cell::new(f64::NEG_INFINITY) };
|
||||
/// Consecutive failures, for the backoff exponent.
|
||||
static SSE_FAILURES: Cell<u32> = const { Cell::new(0) };
|
||||
/// One console warning per degradation, not one per retry.
|
||||
static SSE_WARNED: Cell<bool> = const { Cell::new(false) };
|
||||
}
|
||||
|
||||
/// Open the daemon's SSE channel if it is not already open.
|
||||
///
|
||||
/// The stream carries the same `{"version":N,"collabSeq":M}` payload the
|
||||
/// version poll returns, so it feeds the identical latch and changes only
|
||||
/// *when* a change is noticed — push instead of up to one poll interval later.
|
||||
/// Everything downstream is unchanged, which is what makes losing the stream a
|
||||
/// slowdown rather than a failure.
|
||||
fn ensure_event_stream(base: &str) {
|
||||
if EVENT_STREAM.with(|slot| slot.borrow().is_some()) {
|
||||
return;
|
||||
}
|
||||
if now_ms_f64() < SSE_RETRY_AT_MS.get() {
|
||||
return;
|
||||
}
|
||||
let Ok(stream) = web_sys::EventSource::new(&format!("{base}/api/mcp/events")) else {
|
||||
note_stream_failure();
|
||||
return;
|
||||
};
|
||||
|
||||
use wasm_bindgen::closure::Closure;
|
||||
use wasm_bindgen::JsCast;
|
||||
|
||||
let on_message =
|
||||
Closure::<dyn FnMut(web_sys::MessageEvent)>::new(move |event: web_sys::MessageEvent| {
|
||||
if let Some(payload) = event.data().as_string() {
|
||||
// A live stream proves the daemon is reachable, so a past
|
||||
// failure should not keep throttling reconnects.
|
||||
SSE_FAILURES.set(0);
|
||||
SSE_WARNED.set(false);
|
||||
note_version_probe(&payload);
|
||||
}
|
||||
});
|
||||
stream.set_onmessage(Some(on_message.as_ref().unchecked_ref()));
|
||||
on_message.forget();
|
||||
|
||||
let on_error = Closure::<dyn FnMut(web_sys::Event)>::new(move |_event: web_sys::Event| {
|
||||
// `EventSource` reconnects on its own, but it does so forever and
|
||||
// silently against a daemon that has gone away. Closing it and owning
|
||||
// the backoff keeps the failure visible in one place and bounded.
|
||||
close_event_stream();
|
||||
note_stream_failure();
|
||||
});
|
||||
stream.set_onerror(Some(on_error.as_ref().unchecked_ref()));
|
||||
on_error.forget();
|
||||
|
||||
EVENT_STREAM.with(|slot| *slot.borrow_mut() = Some(stream));
|
||||
}
|
||||
|
||||
fn close_event_stream() {
|
||||
EVENT_STREAM.with(|slot| {
|
||||
if let Some(stream) = slot.borrow_mut().take() {
|
||||
stream.close();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// Record a dropped stream and arm the next reconnect.
|
||||
fn note_stream_failure() {
|
||||
let failures = SSE_FAILURES.get().saturating_add(1);
|
||||
SSE_FAILURES.set(failures);
|
||||
let backoff = (SSE_BASE_RETRY_MS * 2f64.powi(failures.min(5) as i32 - 1)).min(SSE_MAX_RETRY_MS);
|
||||
SSE_RETRY_AT_MS.set(now_ms_f64() + backoff);
|
||||
// One warning per degradation. A daemon that is simply gone would
|
||||
// otherwise fill the console with an identical line every backoff.
|
||||
if !SSE_WARNED.replace(true) {
|
||||
web_sys::console::warn_1(&wasm_bindgen::JsValue::from_str(
|
||||
"[op-collab] event stream unavailable; falling back to polling",
|
||||
));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -297,6 +297,7 @@ fn poll_version<C: RepaintContext + 'static>(
|
|||
sync.borrow_mut().gate.note_conflict(v); // observer latch reports; user decides again
|
||||
return;
|
||||
}
|
||||
maybe_auto_resolve_conflict_in_session(inner, sync, pair);
|
||||
if !sync.borrow().gate.pull_allowed(pair) {
|
||||
return;
|
||||
}
|
||||
|
|
@ -339,6 +340,63 @@ fn poll_version<C: RepaintContext + 'static>(
|
|||
let _ = live_sync::get(&format!("{base}/api/mcp/version"), on_version);
|
||||
}
|
||||
|
||||
/// Auto-resolve a latched push conflict, but only inside a live session.
|
||||
///
|
||||
/// A conflict closes both the pull and the push gate and is cleared by exactly
|
||||
/// two things, neither reachable from this tick: the VS Code host's explicit
|
||||
/// `resolve-conflict`, and a successful re-push. A plain browser tab has
|
||||
/// neither, so a single 409 wedges live sync for the rest of the page's life.
|
||||
///
|
||||
/// Auto-accepting the remote is only safe when the *server* is the authority
|
||||
/// and the user's dropped work is recoverable — which is exactly what an
|
||||
/// `Active` collaboration session provides:
|
||||
///
|
||||
/// * the daemon's document is the sequenced truth every peer already sees, so
|
||||
/// there is nothing for this tab to "win" by holding its copy back;
|
||||
/// * the runtime projects a rejected local edit into
|
||||
/// `collab.discarded_edit`, and the panel offers to replay it, so accepting
|
||||
/// the remote does not silently destroy the edit that lost.
|
||||
///
|
||||
/// Outside a session neither holds — the daemon is a peer, not an authority,
|
||||
/// and nothing preserves the losing edit — so the latch stays and the existing
|
||||
/// explicit-resolution semantics are untouched. That is the difference between
|
||||
/// recovering automatically and quietly overwriting a user's unpushed work.
|
||||
fn maybe_auto_resolve_conflict_in_session<C: RepaintContext + 'static>(
|
||||
inner: &Rc<RefCell<C>>,
|
||||
sync: &SharedSync,
|
||||
pair: (u64, u64),
|
||||
) {
|
||||
let has_conflict = sync.borrow().gate.conflict().is_some();
|
||||
let phase = inner
|
||||
.try_borrow()
|
||||
.map(|context| context.host().editor_state().editor_ui.collab.phase)
|
||||
.unwrap_or(op_editor_core::CollabConnectionPhase::Idle);
|
||||
if !auto_resolve_is_safe(has_conflict, phase) {
|
||||
return;
|
||||
}
|
||||
// Re-opens the pull for THIS pair only; the resolving pull's apply calls
|
||||
// `note_synced`, which is what finally clears the baseline and reopens the
|
||||
// push side too.
|
||||
if let Ok(mut sync) = sync.try_borrow_mut() {
|
||||
sync.gate.resolve_accept_remote(pair);
|
||||
}
|
||||
}
|
||||
|
||||
/// The safety decision behind [`maybe_auto_resolve_conflict_in_session`],
|
||||
/// separated so it can be tested without a live shell.
|
||||
///
|
||||
/// `Active` and nothing else. Every other phase — including `Reconnecting` and
|
||||
/// `ReadOnly`, which look session-ish — either has no authoritative server
|
||||
/// document to accept or no `discarded_edit` projection to recover the losing
|
||||
/// edit from, so auto-accepting there would be the silent data loss this is
|
||||
/// specifically avoiding.
|
||||
const fn auto_resolve_is_safe(
|
||||
has_conflict: bool,
|
||||
phase: op_editor_core::CollabConnectionPhase,
|
||||
) -> bool {
|
||||
has_conflict && matches!(phase, op_editor_core::CollabConnectionPhase::Active)
|
||||
}
|
||||
|
||||
/// Apply a `GET /api/mcp/document` response to the live shell.
|
||||
/// `WebSyncClient::sync_with_editor_meta` runs the apply closure only for a
|
||||
/// newer document and commits that exact version only when the closure returns
|
||||
|
|
@ -723,3 +781,49 @@ mod tests {
|
|||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod auto_resolve_tests {
|
||||
use super::auto_resolve_is_safe;
|
||||
use op_editor_core::CollabConnectionPhase;
|
||||
|
||||
#[test]
|
||||
fn a_conflict_inside_an_active_session_resolves_itself() {
|
||||
assert!(auto_resolve_is_safe(true, CollabConnectionPhase::Active));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn no_conflict_means_nothing_to_resolve() {
|
||||
for phase in [
|
||||
CollabConnectionPhase::Idle,
|
||||
CollabConnectionPhase::Active,
|
||||
CollabConnectionPhase::ReadOnly,
|
||||
] {
|
||||
assert!(!auto_resolve_is_safe(false, phase));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn every_non_active_phase_keeps_the_latch() {
|
||||
// Outside an Active session there is no authoritative server document
|
||||
// to accept and no `discarded_edit` projection to recover the losing
|
||||
// edit from, so auto-accepting would silently destroy unpushed work.
|
||||
// `Reconnecting` and `ReadOnly` look session-ish and are deliberately
|
||||
// included: neither can sequence an edit.
|
||||
for phase in [
|
||||
CollabConnectionPhase::Idle,
|
||||
CollabConnectionPhase::Starting,
|
||||
CollabConnectionPhase::Discovering,
|
||||
CollabConnectionPhase::Joining,
|
||||
CollabConnectionPhase::Authenticating,
|
||||
CollabConnectionPhase::Reconnecting,
|
||||
CollabConnectionPhase::ReadOnly,
|
||||
CollabConnectionPhase::Ended,
|
||||
] {
|
||||
assert!(
|
||||
!auto_resolve_is_safe(true, phase),
|
||||
"{phase:?} must keep the existing explicit-resolution semantics"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -329,6 +329,12 @@ pub struct WidgetHost {
|
|||
/// nodes never collide on the same key. Matches the native
|
||||
/// host's allocator.
|
||||
pub(in crate::widget_host) next_node_id: u64,
|
||||
/// Owner-assigned id allocator, set while a collaboration session is live.
|
||||
///
|
||||
/// The protocol replays node ids verbatim, so a peer minting from the
|
||||
/// private `next_node_id` counter would eventually agree on an id for two
|
||||
/// different nodes. `None` restores the standalone policy.
|
||||
pub(in crate::widget_host) collab_id_allocator: Option<op_editor_core::DocumentIdAllocator>,
|
||||
/// Whether the shift key is currently held. The DOM listener
|
||||
/// updates this from every keyboard / mouse event so apply_press
|
||||
/// can branch on shift+click for multi-select. Matches the
|
||||
|
|
|
|||
|
|
@ -50,6 +50,7 @@ impl WidgetHost {
|
|||
node_drag: None,
|
||||
option_drag_source_ids: Vec::new(),
|
||||
next_node_id: 100,
|
||||
collab_id_allocator: None,
|
||||
shift_held: false,
|
||||
alt_held: false,
|
||||
now_ms: 0,
|
||||
|
|
@ -235,3 +236,53 @@ impl WidgetHost {
|
|||
Some((f64::from(point.x), f64::from(point.y)))
|
||||
}
|
||||
}
|
||||
|
||||
impl WidgetHost {
|
||||
/// Switch every collaboration-supported creation path to one owner-assigned
|
||||
/// namespace, resuming above the ids already in the document.
|
||||
pub(crate) fn enable_collaboration_ids(
|
||||
&mut self,
|
||||
namespace: op_editor_core::PeerNamespace,
|
||||
) -> Result<(), op_editor_core::IdAllocError> {
|
||||
self.collab_id_allocator = Some(
|
||||
op_editor_core::DocumentIdAllocator::namespaced_for_document(
|
||||
&self.editor_state.doc,
|
||||
namespace,
|
||||
)?,
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Return to the standalone `n{counter}` allocation policy.
|
||||
pub(crate) fn disable_collaboration_ids(&mut self) {
|
||||
self.collab_id_allocator = None;
|
||||
if let Ok(next) = op_editor_core::next_sequential_counter(&self.editor_state.doc) {
|
||||
self.next_node_id = self.next_node_id.max(next);
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether creation currently mints from an owner-assigned namespace.
|
||||
pub(crate) fn collaboration_ids_enabled(&self) -> bool {
|
||||
self.collab_id_allocator.is_some()
|
||||
}
|
||||
|
||||
/// Surface an exhausted / invalid namespace as a panel notice.
|
||||
///
|
||||
/// An allocation failure during a session is not something the user can act
|
||||
/// on directly, but silently dropping the gesture would look like a broken
|
||||
/// canvas, so it reports through the same notice channel every other
|
||||
/// collaboration refusal uses.
|
||||
pub(in crate::widget_host) fn show_collab_id_error(
|
||||
&mut self,
|
||||
_error: op_editor_core::IdAllocError,
|
||||
) {
|
||||
let now = self.now_ms;
|
||||
self.editor_state.editor_ui.collab.set_notice(
|
||||
op_editor_core::CollabNoticeKind::Reject(
|
||||
op_editor_core::CollabRejectUiCode::ResourceLimit,
|
||||
),
|
||||
now,
|
||||
);
|
||||
self.mark_dirty();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -66,14 +66,32 @@ impl WidgetHost {
|
|||
let pre_create = self.editor_state.snapshot_for_history();
|
||||
let is_text_tool = matches!(tool, Tool::Text);
|
||||
let (init_w, init_h) = initial_size_for_tool(tool);
|
||||
let Some(node_id) = self.editor_state.create_node_for_tool(
|
||||
tool,
|
||||
&mut self.next_node_id,
|
||||
doc_point.x as f64,
|
||||
doc_point.y as f64,
|
||||
init_w,
|
||||
init_h,
|
||||
) else {
|
||||
let created = if let Some(allocator) = self.collab_id_allocator.as_mut() {
|
||||
self.editor_state.create_node_for_tool_with_allocator(
|
||||
tool,
|
||||
allocator,
|
||||
doc_point.x as f64,
|
||||
doc_point.y as f64,
|
||||
init_w,
|
||||
init_h,
|
||||
)
|
||||
} else {
|
||||
Ok(self.editor_state.create_node_for_tool(
|
||||
tool,
|
||||
&mut self.next_node_id,
|
||||
doc_point.x as f64,
|
||||
doc_point.y as f64,
|
||||
init_w,
|
||||
init_h,
|
||||
))
|
||||
};
|
||||
let Some(node_id) = (match created {
|
||||
Ok(id) => id,
|
||||
Err(error) => {
|
||||
self.show_collab_id_error(error);
|
||||
return true;
|
||||
}
|
||||
}) else {
|
||||
return false;
|
||||
};
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue