Merge pull request #80 from levineuwirth/session-gpu-attach-robustness
fix(gpu): attach robustness — non-CRDT error, bounded queue, clamped dropdown (F-003/F-008/F-007)
This commit is contained in:
GitHub
commit
3898f5b721
|
|
@ -0,0 +1,189 @@
|
|||
# GPU attach robustness — framing + as-built
|
||||
|
||||
Three audit findings on the `pmacs-gpu` attach path, taken as one arc
|
||||
because they share a surface (the frontend↔daemon connection) and a
|
||||
theme — the GPU frontend fails *quietly* where it should fail *loudly* or
|
||||
*bounded*. None touches the wire protocol or the daemon; all three live in
|
||||
`pmacs-gpu` (`attach.rs` + `main.rs`).
|
||||
|
||||
- **F-003** — attach against a daemon built without `--features crdt`
|
||||
negotiates "successfully", then sits on `(connecting...)` forever
|
||||
because no `BufferSnapshot` ever arrives. Silent, confusing.
|
||||
- **F-008** — the outbound `FrontendEvent` queue is an unbounded
|
||||
`mpsc::channel`. A stalled daemon lets the UI thread enqueue without
|
||||
backpressure: unbounded memory, and a backlog of stale viewport/pointer
|
||||
traffic replayed on recovery.
|
||||
- **F-007** — the minibuffer completion dropdown grows upward from the
|
||||
status band by `n * row_height` with no clamp; on a very short window
|
||||
it renders above `y=0` and the selected row can be off-screen.
|
||||
|
||||
## What the recon established
|
||||
|
||||
- **F-003 needs no protocol change.** `connect()` (`attach.rs:95`) reads
|
||||
`Hello`, checks `protocol_version`, then sends `AttachRequest`
|
||||
**fire-and-forget** — it never inspects what the daemon granted. But
|
||||
`Hello.instance_capabilities` *already* carries `crdt_replica`,
|
||||
`semantic_render`, and `multi_frontend` (the daemon advertises them; on
|
||||
a non-CRDT build they default `false`). The GPU can read the field it
|
||||
was already sent and reject the attach itself. The call site
|
||||
(`main.rs:751`) already renders a fatal on-screen line
|
||||
(`set_text("(attach failed; see stderr)")`) — the rendering path exists.
|
||||
- **F-008's flood sources are `Viewport` and `Pointer{Drag}`.** Viewport
|
||||
fires on every wheel/scrub/resize/edge-scroll; `Drag` on every
|
||||
cursor-move-with-button. The *other* pointer kinds — `Down`, `Up`,
|
||||
`DoubleDown`, `TripleDown`, `Context` — are discrete clicks that are
|
||||
causally ordered (a `Down`/`Up` pair *is* a selection gesture; `Context`
|
||||
opens a menu). `Key`, `CrdtOp`, `Paste`, `MenuPointer` are all
|
||||
order-critical too — `CrdtOp` especially, since the GPU applies it to
|
||||
its local Loro replica optimistically, so a dropped one desyncs.
|
||||
- **F-007 already clips.** The dropdown's `TextArea` sets
|
||||
`bounds.bottom = band_top` (`main.rs:3946`); glyphon clips glyphs to the
|
||||
rect. So the fix is geometry-only: shape all candidates once, then
|
||||
scroll + clamp the visible window at render time — no re-shape on
|
||||
resize, and the common (fits) case stays byte-identical.
|
||||
|
||||
## The rules
|
||||
|
||||
**Q#AR1 — F-003: detect the capability mismatch client-side, fail loud.**
|
||||
After the version check, read `hello.instance_capabilities`; collect any
|
||||
of `{multi_frontend, crdt_replica, semantic_render}` the daemon does *not*
|
||||
advertise. If the set is non-empty, return a new
|
||||
`AttachClientError::CapabilityMismatch { missing }` instead of proceeding.
|
||||
`Display` spells out the fix (start the daemon built with `--features
|
||||
crdt`); the call site renders a concise, actionable on-screen line
|
||||
(`window_status()` per variant) rather than the generic "see stderr". No
|
||||
`AttachResponse`, no daemon `Goodbye` — the daemon already told us in
|
||||
`Hello`; we just have to read it. (This is *better* than the audit's
|
||||
proposed daemon-side rejection: the daemon needn't know what each frontend
|
||||
requires; the frontend checks the advertised caps against its own needs.)
|
||||
|
||||
**Q#AR2 — F-008: bound the queue and trailing-coalesce the floods.**
|
||||
Replace the unbounded `mpsc` with an `Arc<(Mutex<Outbox>, Condvar)>` the
|
||||
writer thread drains. `Outbox` is a `VecDeque<FrontendEvent>` + a `closed`
|
||||
flag. Enqueue policy:
|
||||
|
||||
- **Coalesceable** = `Viewport` and `Pointer{kind: Drag}`. When enqueuing
|
||||
one, if the queue's **tail** is the same coalesceable kind, *replace*
|
||||
it; else append. This collapses a run of scroll/drag spam to O(1)
|
||||
**without reordering** across an intervening click or key — a `Down,
|
||||
Drag, Drag, Drag, Up` gesture becomes `Down, Drag(latest), Up`, never
|
||||
`Down, Up, Drag`.
|
||||
- **Everything else is lossless** — appended in order. If appending would
|
||||
exceed `OUTBOX_MAX`, set `closed` and return `Err` (fail-fast): a
|
||||
daemon so stalled that thousands of keys/ops piled up is dead, and a
|
||||
clean disconnect → reconnect → fresh snapshot is more correct than
|
||||
silently dropping a `CrdtOp` (which desyncs the optimistic replica).
|
||||
|
||||
The writer waits on the condvar, drains the whole batch into a local
|
||||
`Vec` (releasing the lock before any blocking socket write), writes each,
|
||||
and exits on write error or `closed`. `send_event` keeps its
|
||||
`Result<(), TransportError>` signature (callers already log + continue).
|
||||
|
||||
**Q#AR3 — F-007: window the dropdown to what fits, keep the selection
|
||||
visible.** One helper `mb_visible_window() -> Option<(first, count)>`:
|
||||
`count = min(n, floor(band_top / MB_DROP_ROW_HEIGHT))` (≥1), and `first`
|
||||
scrolls so `mb.selected` stays in `[first, first+count)`. The three
|
||||
consumers share it — `mb_dropdown_rect` sizes the box by `count` (so
|
||||
`top_y ≥ 0` by construction), `mb_dropdown_vertex_bytes` draws `count`
|
||||
rows and offsets the selection highlight by `first`, and the text
|
||||
`TextArea` sets `top = top_y - first*row` so line `first` lands at the box
|
||||
top while `bounds.top = top_y` clips the rows scrolled above. When the
|
||||
whole list fits (`count == n`, `first == 0`) every value is identical to
|
||||
today — no regression on the common path.
|
||||
|
||||
## Categorical bets
|
||||
|
||||
- **Read what you're already sent (F-003).** The daemon advertises its
|
||||
capabilities in `Hello`; a client that ignores them and waits for a
|
||||
snapshot that structurally can't come is the bug. No new wire state.
|
||||
- **Coalesce by tail-replacement, not a side slot (F-008).** A separate
|
||||
"latest viewport/pointer" slot would reorder those events relative to
|
||||
the ordered stream and break the `Down/Drag/Up` causal chain.
|
||||
Tail-replacement collapses exactly the consecutive runs that flood,
|
||||
and only those, preserving order by construction.
|
||||
- **Never silently drop a lossless event (F-008).** `CrdtOp` is applied
|
||||
optimistically; dropping it desyncs. Under true overflow, fail-fast to a
|
||||
reconnect (clean resync) beats a silent divergence.
|
||||
- **Geometry-only for F-007.** glyphon already clips; shaping all
|
||||
candidates once and scrolling at render keeps resize free and the common
|
||||
path unchanged.
|
||||
|
||||
## Validation implication
|
||||
|
||||
F-003 and F-007 are eyeball-confirmable locally (this box has a Vulkan
|
||||
adapter + a CRDT daemon): attach to a non-CRDT daemon → see the actionable
|
||||
banner instead of a hang; shrink the window under an open completion →
|
||||
dropdown stays on-screen with the selection visible. F-008's coalescing
|
||||
and bound are unit-tested (tail-replacement collapses runs, preserves
|
||||
order across a click, caps at `OUTBOX_MAX`); its behavior under a real
|
||||
stalled daemon is not easily reproduced in a test and rides on the logic
|
||||
proof + the existing reader-driven disconnect path.
|
||||
|
||||
## As-built
|
||||
|
||||
Landed as framed; all three in `pmacs-gpu`, no protocol/daemon change.
|
||||
|
||||
- **F-003** (`attach.rs`): new `AttachClientError::CapabilityMismatch {
|
||||
missing }`; a free `missing_capabilities(&InstanceCapabilities)` checks
|
||||
`multi_frontend`/`crdt_replica`/`semantic_render` against
|
||||
`Hello.instance_capabilities` right after the version check and returns
|
||||
the error before sending `AttachRequest`. `window_status()` gives the
|
||||
call site (`main.rs`) a concise in-window line ("daemon lacks CRDT
|
||||
support — restart it built with `--features crdt`") while `Display`
|
||||
keeps the full detail for stderr. 3 unit tests.
|
||||
- **F-008** (`attach.rs`): the unbounded `mpsc` became
|
||||
`Arc<(Mutex<Outbox>, Condvar)>`. `Outbox::enqueue` tail-replaces a
|
||||
same-kind `Viewport`/`Pointer{Drag}` (coalesce), appends everything else
|
||||
lossless, and on a lossless append past `OUTBOX_MAX` sets `closed` +
|
||||
returns `false` (fail-fast). The writer waits on the condvar, `mem::take`s
|
||||
the whole batch, and writes with the lock released.
|
||||
**Fail-fast is a real teardown, not just a flag** (review follow-up):
|
||||
closing the outbox alone left the reader blocked on its still-open
|
||||
socket clone, so the optimistic edit whose `send_crdt_op` failed was
|
||||
applied locally, logged, and forgotten — silent divergence, the exact
|
||||
stalled-daemon case F-008 targets. Now a `shutdown_handle` clone is
|
||||
`shutdown(Both)` whenever the outbox closes (overflow in `send_event`,
|
||||
or a writer write error): the reader wakes with EOF and fires the
|
||||
existing `Disconnected` path (`(daemon disconnected)` in the window),
|
||||
and the daemon sees the half-close. 6 unit tests (coalesce-to-latest,
|
||||
clicks keep order, same-kind-tail only, overflow closes, coalescing is
|
||||
uncapped, and a socketpair test asserting a closed outbox drives the
|
||||
peer to EOF).
|
||||
- **F-007** (`main.rs`): a pure `mb_dropdown_window(n, selected, band_top)
|
||||
-> Option<(first, count)>` clamps the row count to what fits above the
|
||||
band (hides entirely when not even one row fits, so `top_y` is never
|
||||
negative) and scrolls to keep the selection visible. The method
|
||||
`mb_visible_window` feeds `mb_dropdown_rect` (box sized by `count`),
|
||||
`mb_dropdown_vertex_bytes` (bg + selection offset by `first`), and the
|
||||
render `TextArea` (`top = top_y - first*row`, existing `bounds` clip the
|
||||
scrolled-out rows). `(0, n)` when the list fits — the common path is
|
||||
byte-identical to before. 1 unit test.
|
||||
|
||||
Validated: `cargo fmt` clean; `clippy -p pmacs-gpu --all-targets` clean;
|
||||
52 pmacs-gpu unit tests pass, including the two headless render tests on
|
||||
this box's Vulkan adapter (`PMACS_REQUIRE_GPU=1`). Divergence from the
|
||||
framing: the F-008 fail-fast needed an actual socket teardown, added as a
|
||||
review follow-up (above); the framing's "clean disconnect" was otherwise
|
||||
aspirational.
|
||||
|
||||
**Still needs a human eyeball before merge** (per the validation
|
||||
implication above): attach to a non-CRDT daemon → the actionable banner;
|
||||
shrink the window under an open completion → the dropdown stays on-screen
|
||||
with the selection visible; and a normal attach still renders/resizes
|
||||
(the queue rewrite is on the live write path).
|
||||
|
||||
## Deferred (named)
|
||||
|
||||
- **F-008 auto-reconnect / resync.** Overflow-close now actively tears the
|
||||
session down and shows `(daemon disconnected)` (the review follow-up
|
||||
above). What's still deferred is *recovery*: `pmacs-gpu` has no
|
||||
auto-reconnect, so the diverged optimistic replica is reconciled only by
|
||||
a fresh `BufferSnapshot` on a manual re-attach. A "daemon not keeping up
|
||||
— reconnecting…" banner + automatic re-attach belongs with the reconnect
|
||||
thread ([[attach_reconnect]] already exists daemon-side for the TUI).
|
||||
- **F-003 capability renegotiation.** We reject on missing caps; a
|
||||
friendlier flow would offer to relaunch the daemon with `crdt`. Out of
|
||||
scope — the frontend can't manage the daemon's lifecycle.
|
||||
- **F-007 pointer hit-testing.** The minibuffer dropdown is keyboard-only
|
||||
(no pointer hit-test today), so "hit testing uses the same window" is
|
||||
vacuous now; revisit if the dropdown becomes clickable.
|
||||
|
|
@ -16,9 +16,10 @@
|
|||
//! wakes the main loop so the frame logic can apply the message and
|
||||
//! redraw.
|
||||
|
||||
use std::collections::VecDeque;
|
||||
use std::os::unix::net::UnixStream;
|
||||
use std::path::Path;
|
||||
use std::sync::mpsc;
|
||||
use std::sync::{Arc, Condvar, Mutex};
|
||||
use std::thread;
|
||||
|
||||
use pmacs_protocol::{
|
||||
|
|
@ -43,6 +44,27 @@ pub enum AttachClientError {
|
|||
Handshake(TransportError),
|
||||
/// Server's `protocol_version` is outside `SUPPORTED_PROTOCOL_VERSIONS`.
|
||||
VersionMismatch { server: u32, client: u32 },
|
||||
/// The daemon's `Hello.instance_capabilities` doesn't advertise a
|
||||
/// capability a semantic frontend needs (audit F-003). Most commonly
|
||||
/// the daemon was built without `--features crdt`, so `crdt_replica`
|
||||
/// / `semantic_render` are `false`: negotiation would "succeed" but no
|
||||
/// `BufferSnapshot` ever arrives and the window sits on
|
||||
/// `(connecting...)` forever. We reject up front instead.
|
||||
CapabilityMismatch { missing: Vec<&'static str> },
|
||||
}
|
||||
|
||||
impl AttachClientError {
|
||||
/// A short, actionable line to render *in the window* (the user may
|
||||
/// never see stderr). The `Display` impl carries the full detail for
|
||||
/// logs; this is the one-liner for the failed-attach placeholder.
|
||||
pub fn window_status(&self) -> String {
|
||||
match self {
|
||||
Self::CapabilityMismatch { .. } => {
|
||||
"daemon lacks CRDT support — restart it built with `--features crdt`".to_owned()
|
||||
}
|
||||
_ => "(attach failed; see stderr)".to_owned(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Display for AttachClientError {
|
||||
|
|
@ -55,12 +77,35 @@ impl std::fmt::Display for AttachClientError {
|
|||
"daemon protocol version {server} not in client's supported set (this client = \
|
||||
{client}, supports {SUPPORTED_PROTOCOL_VERSIONS:?})"
|
||||
),
|
||||
Self::CapabilityMismatch { missing } => write!(
|
||||
f,
|
||||
"daemon does not advertise required capabilities {missing:?} — start the daemon \
|
||||
built with the `crdt` feature (it advertises `crdt_replica` / `semantic_render` \
|
||||
only on CRDT builds; without them no BufferSnapshot is ever sent)"
|
||||
),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for AttachClientError {}
|
||||
|
||||
/// The capabilities a semantic `pmacs-gpu` frontend requires the daemon to
|
||||
/// advertise in `Hello.instance_capabilities`, and which of them this
|
||||
/// daemon is missing (audit F-003). Empty ⇒ the attach can proceed.
|
||||
fn missing_capabilities(caps: &pmacs_protocol::InstanceCapabilities) -> Vec<&'static str> {
|
||||
let mut missing = Vec::new();
|
||||
if !caps.multi_frontend {
|
||||
missing.push("multi_frontend");
|
||||
}
|
||||
if !caps.crdt_replica {
|
||||
missing.push("crdt_replica");
|
||||
}
|
||||
if !caps.semantic_render {
|
||||
missing.push("semantic_render");
|
||||
}
|
||||
missing
|
||||
}
|
||||
|
||||
/// One decoded event forwarded from the reader thread to the main
|
||||
/// loop. `Message` carries the entire `InstanceMessage`; `Disconnected`
|
||||
/// fires once when the reader thread exits (clean EOF or transport
|
||||
|
|
@ -74,6 +119,82 @@ pub enum AttachEvent {
|
|||
Disconnected(String),
|
||||
}
|
||||
|
||||
/// Hard cap on queued **lossless** outbound events (audit F-008). A
|
||||
/// daemon so stalled that this many keys / CRDT ops / pastes pile up is
|
||||
/// effectively dead; past it we fail fast (see [`Outbox::enqueue`]) rather
|
||||
/// than grow memory without bound. Coalesceable events (viewport / drag)
|
||||
/// never count against it — they collapse into the queue tail. 8192 is far
|
||||
/// above any human-paced burst, so it only trips on a genuine stall.
|
||||
const OUTBOX_MAX: usize = 8192;
|
||||
|
||||
/// The kind tag of the two high-frequency events coalesced by
|
||||
/// tail-replacement (F-008): scroll `Viewport`s and `Pointer` **drags**.
|
||||
/// `None` marks a *lossless* event — `Key`, `CrdtOp`, `Paste`,
|
||||
/// `MenuPointer`, and the discrete pointer clicks (`Down` / `Up` /
|
||||
/// `DoubleDown` / `TripleDown` / `Context`) — which are ordered and never
|
||||
/// dropped. Coalescing a drag is safe only against a *same-kind tail*, so
|
||||
/// a `Down, Drag, Drag, Up` gesture keeps its `Down`/`Up` ordering.
|
||||
fn coalesce_kind(event: &FrontendEvent) -> Option<u8> {
|
||||
match event {
|
||||
FrontendEvent::Viewport { .. } => Some(0),
|
||||
FrontendEvent::Pointer {
|
||||
kind: PointerKind::Drag,
|
||||
..
|
||||
} => Some(1),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Bounded, coalescing outbound queue drained by the writer thread
|
||||
/// (audit F-008). Replaces the previous unbounded `mpsc::channel`, which
|
||||
/// let a stalled daemon grow memory without limit and replay a backlog of
|
||||
/// stale viewport/pointer traffic on recovery.
|
||||
struct Outbox {
|
||||
queue: VecDeque<FrontendEvent>,
|
||||
/// Set once the writer gives up (socket error) or a lossless overflow
|
||||
/// trips fail-fast. Further [`Outbox::enqueue`] calls return `false`.
|
||||
closed: bool,
|
||||
}
|
||||
|
||||
impl Outbox {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
queue: VecDeque::new(),
|
||||
closed: false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Apply the F-008 enqueue policy. Returns `false` when the event
|
||||
/// can't be accepted — the outbox is `closed`, or a lossless append
|
||||
/// would exceed [`OUTBOX_MAX`] (which also *sets* `closed`: a clean
|
||||
/// disconnect → reconnect → fresh snapshot is more correct than
|
||||
/// silently dropping a `CrdtOp` and desyncing the optimistic replica).
|
||||
///
|
||||
/// A coalesceable event (viewport / drag) whose kind matches the queue
|
||||
/// **tail** *replaces* it — collapsing a scroll or drag flood to O(1)
|
||||
/// without reordering across an intervening click or key.
|
||||
fn enqueue(&mut self, event: FrontendEvent) -> bool {
|
||||
if self.closed {
|
||||
return false;
|
||||
}
|
||||
if let Some(kind) = coalesce_kind(&event)
|
||||
&& self.queue.back().and_then(coalesce_kind) == Some(kind)
|
||||
{
|
||||
*self
|
||||
.queue
|
||||
.back_mut()
|
||||
.expect("tail present when back() matched") = event;
|
||||
return true;
|
||||
}
|
||||
if self.queue.len() >= OUTBOX_MAX {
|
||||
self.closed = true;
|
||||
return false;
|
||||
}
|
||||
self.queue.push_back(event);
|
||||
true
|
||||
}
|
||||
}
|
||||
|
||||
/// Connect, handshake, and spawn the reader thread.
|
||||
///
|
||||
/// Returns once the handshake has completed and the reader thread is
|
||||
|
|
@ -112,20 +233,23 @@ pub fn connect(
|
|||
hello.protocol_version, hello.instance_identity.pmacs_version
|
||||
);
|
||||
|
||||
// Capability gate (audit F-003). The daemon advertises what it can do
|
||||
// in `Hello.instance_capabilities`; a semantic frontend needs
|
||||
// `multi_frontend` + `crdt_replica` + `semantic_render`. A daemon
|
||||
// built without `--features crdt` advertises those as `false`: the
|
||||
// handshake would otherwise "succeed", but no `BufferSnapshot` ever
|
||||
// arrives and the window sits on `(connecting...)` forever. Reject up
|
||||
// front with an actionable error rather than hanging silently. No
|
||||
// `AttachResponse` round-trip is needed — the daemon already told us
|
||||
// in `Hello`; we just have to check it against what we require.
|
||||
let missing = missing_capabilities(&hello.instance_capabilities);
|
||||
if !missing.is_empty() {
|
||||
return Err(AttachClientError::CapabilityMismatch { missing });
|
||||
}
|
||||
|
||||
// AttachRequest — declare the capabilities a semantic frontend
|
||||
// needs. `multi_frontend` is included because the existing daemon
|
||||
// gates `crdt_replica` behind it (M10.x dependency).
|
||||
//
|
||||
// **Daemon requirement**: the daemon must be built with the
|
||||
// `crdt` feature (`cargo run --features crdt --bin pmacs --
|
||||
// --daemon ...`). Without it the daemon's
|
||||
// `InstanceCapabilities::default` returns `crdt_replica: false`,
|
||||
// negotiation succeeds but no `BufferSnapshot` ever arrives, and
|
||||
// the `pmacs-gpu` window sits on `(connecting...)` forever. This
|
||||
// surfaced as a session-3 finding when manually validating the
|
||||
// attach loop; classified as small under rule (iii) — recorded
|
||||
// here so the next person attaching against a non-crdt daemon
|
||||
// recognizes the symptom immediately.
|
||||
let req = AttachRequest {
|
||||
protocol_version: hello.protocol_version,
|
||||
frontend_capabilities: FrontendCapabilities {
|
||||
|
|
@ -152,8 +276,19 @@ pub fn connect(
|
|||
// buffer state — safe to read on one clone while the other writes
|
||||
// (the FD is full-duplex).
|
||||
let mut read_stream = stream.try_clone().map_err(AttachClientError::Connect)?;
|
||||
// A third clone kept solely to *shut the socket down* (F-008). When the
|
||||
// outbox closes — a lossless overflow, or a writer write error — we
|
||||
// `shutdown(Both)` so the reader (blocked in `read_message` on its own
|
||||
// clone) wakes with EOF and fires `Disconnected`. That routes the
|
||||
// stall into the existing visible teardown instead of leaving the GPU
|
||||
// showing locally-applied edits the daemon never received. Clones share
|
||||
// the socket's file description, so a shutdown on any of them affects
|
||||
// all — and it half-closes toward the daemon so it sees the departure.
|
||||
let shutdown_handle = stream.try_clone().map_err(AttachClientError::Connect)?;
|
||||
let write_stream = stream;
|
||||
let (writer_tx, writer_rx) = mpsc::channel::<FrontendEvent>();
|
||||
// Bounded, coalescing outbound queue (F-008) shared with the writer
|
||||
// thread; the `Condvar` wakes the writer when the UI thread enqueues.
|
||||
let outbox = Arc::new((Mutex::new(Outbox::new()), Condvar::new()));
|
||||
|
||||
// Reader thread. Each iteration: block on read_message, decode,
|
||||
// forward via the event-loop proxy. Exits cleanly on EOF / any
|
||||
|
|
@ -185,22 +320,45 @@ pub fn connect(
|
|||
|
||||
// Writer thread. Socket writes can block when the daemon falls
|
||||
// behind; doing them here keeps keyboard input, redraws, and
|
||||
// message application off that backpressure path.
|
||||
// message application off that backpressure path. It waits on the
|
||||
// outbox condvar, takes the whole pending batch, and *releases the
|
||||
// lock before* the blocking writes so the UI thread can keep
|
||||
// enqueueing (and coalescing) meanwhile.
|
||||
let writer_outbox = Arc::clone(&outbox);
|
||||
thread::Builder::new()
|
||||
.name("pmacs-gpu attach writer".into())
|
||||
.spawn(move || {
|
||||
let mut write_stream = write_stream;
|
||||
while let Ok(event) = writer_rx.recv() {
|
||||
if let Err(e) = write_message(&mut write_stream, &event) {
|
||||
eprintln!("pmacs-gpu: attach writer stopped: {e}");
|
||||
return;
|
||||
let (lock, cvar) = &*writer_outbox;
|
||||
loop {
|
||||
let batch = {
|
||||
let mut ob = lock.lock().expect("outbox lock");
|
||||
while ob.queue.is_empty() && !ob.closed {
|
||||
ob = cvar.wait(ob).expect("outbox condvar wait");
|
||||
}
|
||||
if ob.queue.is_empty() && ob.closed {
|
||||
return;
|
||||
}
|
||||
std::mem::take(&mut ob.queue)
|
||||
};
|
||||
for event in batch {
|
||||
if let Err(e) = write_message(&mut write_stream, &event) {
|
||||
eprintln!("pmacs-gpu: attach writer stopped: {e}");
|
||||
lock.lock().expect("outbox lock").closed = true;
|
||||
// Wake the reader (and signal the daemon) so the
|
||||
// session tears down visibly (F-008) rather than
|
||||
// leaving the reader blocked on a dead socket.
|
||||
let _ = write_stream.shutdown(std::net::Shutdown::Both);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
.expect("spawn attach writer thread");
|
||||
|
||||
Ok(AttachClient {
|
||||
writer_tx,
|
||||
outbox,
|
||||
shutdown_handle,
|
||||
frontend_id: hello.assigned_frontend_id,
|
||||
server_protocol_version: hello.protocol_version,
|
||||
})
|
||||
|
|
@ -209,7 +367,14 @@ pub fn connect(
|
|||
/// Handle the main loop keeps after `connect` returns. It queues
|
||||
/// `FrontendEvent`s for the attach writer thread.
|
||||
pub struct AttachClient {
|
||||
writer_tx: mpsc::Sender<FrontendEvent>,
|
||||
/// Bounded, coalescing outbound queue (F-008) drained by the writer
|
||||
/// thread. `send_event` locks it, applies the enqueue policy, and
|
||||
/// wakes the writer via the paired `Condvar`.
|
||||
outbox: Arc<(Mutex<Outbox>, Condvar)>,
|
||||
/// Socket clone used only to `shutdown(Both)` when the outbox closes
|
||||
/// (F-008), so the reader wakes and the session tears down visibly
|
||||
/// instead of diverging silently. See [`connect`].
|
||||
shutdown_handle: UnixStream,
|
||||
/// Assigned by the daemon in the `Hello` response. Every
|
||||
/// `FrontendEvent` carries this so the daemon can route input back
|
||||
/// to the per-session `SemanticRenderState`.
|
||||
|
|
@ -324,11 +489,237 @@ impl AttachClient {
|
|||
}
|
||||
|
||||
fn send_event(&self, event: FrontendEvent) -> Result<(), TransportError> {
|
||||
self.writer_tx.send(event).map_err(|_| {
|
||||
TransportError::Io(std::io::Error::new(
|
||||
let (lock, cvar) = &*self.outbox;
|
||||
let mut ob = lock.lock().expect("outbox lock");
|
||||
if ob.enqueue(event) {
|
||||
// Drop the guard before notifying so the woken writer doesn't
|
||||
// immediately re-block on a still-held lock.
|
||||
drop(ob);
|
||||
cvar.notify_one();
|
||||
Ok(())
|
||||
} else {
|
||||
// Refused because the outbox is closed — a lossless overflow
|
||||
// against a stalled daemon (or an earlier writer failure).
|
||||
// Tear the session down actively (F-008): shut the socket so
|
||||
// the reader wakes with EOF and fires `Disconnected`, giving a
|
||||
// visible "(daemon disconnected)" instead of a GPU that keeps
|
||||
// showing optimistic edits the daemon never received. Idempotent
|
||||
// — a second shutdown just returns `NotConnected`, ignored.
|
||||
drop(ob);
|
||||
let _ = self.shutdown_handle.shutdown(std::net::Shutdown::Both);
|
||||
Err(TransportError::Io(std::io::Error::new(
|
||||
std::io::ErrorKind::BrokenPipe,
|
||||
"attach writer thread stopped",
|
||||
))
|
||||
})
|
||||
"attach writer stopped or outbound queue overflowed",
|
||||
)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use pmacs_protocol::InstanceCapabilities;
|
||||
|
||||
fn caps(
|
||||
multi_frontend: bool,
|
||||
crdt_replica: bool,
|
||||
semantic_render: bool,
|
||||
) -> InstanceCapabilities {
|
||||
InstanceCapabilities {
|
||||
multi_frontend,
|
||||
crdt_replica,
|
||||
semantic_render,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn full_crdt_daemon_advertises_everything_required() {
|
||||
// A daemon built with `--features crdt` advertises all three — the
|
||||
// attach proceeds (F-003).
|
||||
assert!(missing_capabilities(&caps(true, true, true)).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn non_crdt_daemon_is_rejected_with_the_missing_caps() {
|
||||
// The exact silent-hang case: no crdt build ⇒ crdt_replica /
|
||||
// semantic_render false. We name them so the error is actionable.
|
||||
let missing = missing_capabilities(&caps(true, false, false));
|
||||
assert_eq!(missing, vec!["crdt_replica", "semantic_render"]);
|
||||
|
||||
// A single-frontend daemon is also unusable for a GPU attach.
|
||||
assert_eq!(
|
||||
missing_capabilities(&caps(false, true, true)),
|
||||
vec!["multi_frontend"]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn capability_mismatch_status_line_is_actionable() {
|
||||
// The in-window line (not stderr) must point at the fix.
|
||||
let err = AttachClientError::CapabilityMismatch {
|
||||
missing: vec!["crdt_replica"],
|
||||
};
|
||||
let status = err.window_status();
|
||||
assert!(
|
||||
status.contains("crdt"),
|
||||
"status should name the fix: {status}"
|
||||
);
|
||||
// Other errors keep the generic placeholder.
|
||||
let other = AttachClientError::VersionMismatch {
|
||||
server: 99,
|
||||
client: 6,
|
||||
};
|
||||
assert_eq!(other.window_status(), "(attach failed; see stderr)");
|
||||
}
|
||||
|
||||
// --- F-008: bounded, coalescing outbox --------------------------------
|
||||
|
||||
fn fe_viewport(generation: u64) -> FrontendEvent {
|
||||
FrontendEvent::Viewport {
|
||||
frontend_id: FrontendId::LOCAL,
|
||||
buffer_id: BufferId::from_raw(1),
|
||||
visible: ByteRange { start: 0, end: 10 },
|
||||
generation,
|
||||
}
|
||||
}
|
||||
|
||||
fn fe_pointer(kind: PointerKind, byte: u64) -> FrontendEvent {
|
||||
FrontendEvent::Pointer {
|
||||
frontend_id: FrontendId::LOCAL,
|
||||
buffer_id: BufferId::from_raw(1),
|
||||
byte,
|
||||
kind,
|
||||
mods: Modifiers::NONE,
|
||||
}
|
||||
}
|
||||
|
||||
fn fe_key(c: char) -> FrontendEvent {
|
||||
FrontendEvent::Key(KeyEvent {
|
||||
frontend_id: FrontendId::LOCAL,
|
||||
key: Key::Char(c),
|
||||
mods: Modifiers::NONE,
|
||||
timestamp_ns: 0,
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn consecutive_viewports_coalesce_to_the_latest() {
|
||||
let mut ob = Outbox::new();
|
||||
assert!(ob.enqueue(fe_viewport(1)));
|
||||
assert!(ob.enqueue(fe_viewport(2)));
|
||||
assert!(ob.enqueue(fe_viewport(3)));
|
||||
// A scroll flood collapses to one — the newest generation.
|
||||
assert_eq!(ob.queue.len(), 1);
|
||||
match &ob.queue[0] {
|
||||
FrontendEvent::Viewport { generation, .. } => assert_eq!(*generation, 3),
|
||||
other => panic!("expected a viewport, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn drags_coalesce_but_clicks_keep_their_order() {
|
||||
let mut ob = Outbox::new();
|
||||
// A press, a run of motion, and a release.
|
||||
ob.enqueue(fe_pointer(PointerKind::Down, 0));
|
||||
ob.enqueue(fe_pointer(PointerKind::Drag, 1));
|
||||
ob.enqueue(fe_pointer(PointerKind::Drag, 2));
|
||||
ob.enqueue(fe_pointer(PointerKind::Drag, 3));
|
||||
ob.enqueue(fe_pointer(PointerKind::Up, 4));
|
||||
// Down, one coalesced Drag (latest byte), Up — order preserved, the
|
||||
// gesture is intact; the drag run collapsed to O(1).
|
||||
assert_eq!(ob.queue.len(), 3);
|
||||
assert!(matches!(
|
||||
&ob.queue[0],
|
||||
FrontendEvent::Pointer {
|
||||
kind: PointerKind::Down,
|
||||
..
|
||||
}
|
||||
));
|
||||
assert!(matches!(
|
||||
&ob.queue[1],
|
||||
FrontendEvent::Pointer {
|
||||
kind: PointerKind::Drag,
|
||||
byte: 3,
|
||||
..
|
||||
}
|
||||
));
|
||||
assert!(matches!(
|
||||
&ob.queue[2],
|
||||
FrontendEvent::Pointer {
|
||||
kind: PointerKind::Up,
|
||||
..
|
||||
}
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn coalescing_only_collapses_a_same_kind_tail() {
|
||||
let mut ob = Outbox::new();
|
||||
// Keys between viewports break the run: nothing coalesces, order
|
||||
// and every lossless event are preserved.
|
||||
ob.enqueue(fe_key('a'));
|
||||
ob.enqueue(fe_viewport(1));
|
||||
ob.enqueue(fe_key('b'));
|
||||
ob.enqueue(fe_viewport(2));
|
||||
assert_eq!(ob.queue.len(), 4);
|
||||
// A drag does not fold into a Down tail (different kind).
|
||||
let mut ob2 = Outbox::new();
|
||||
ob2.enqueue(fe_pointer(PointerKind::Down, 0));
|
||||
ob2.enqueue(fe_pointer(PointerKind::Drag, 1));
|
||||
assert_eq!(ob2.queue.len(), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn lossless_overflow_fails_fast_and_closes() {
|
||||
let mut ob = Outbox::new();
|
||||
for i in 0..OUTBOX_MAX {
|
||||
assert!(ob.enqueue(fe_key('x')), "fill up to the cap (i={i})");
|
||||
}
|
||||
// The cap-crossing lossless event is rejected and the outbox is
|
||||
// now closed — a clean disconnect beats dropping a CrdtOp silently.
|
||||
assert!(!ob.enqueue(fe_key('y')));
|
||||
assert!(ob.closed);
|
||||
// Once closed, everything is refused (including coalesceable kinds).
|
||||
assert!(!ob.enqueue(fe_viewport(1)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn coalescing_does_not_count_against_the_cap() {
|
||||
let mut ob = Outbox::new();
|
||||
// Even a huge scroll flood stays at one queued event, so it can
|
||||
// never trip the overflow fail-fast.
|
||||
for g in 0..(OUTBOX_MAX as u64 * 4) {
|
||||
assert!(ob.enqueue(fe_viewport(g)));
|
||||
}
|
||||
assert_eq!(ob.queue.len(), 1);
|
||||
assert!(!ob.closed);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_closed_outbox_shuts_the_socket_down_to_wake_the_reader() {
|
||||
use std::io::Read;
|
||||
// A socketpair stands in for the daemon connection; `a` is the peer
|
||||
// a blocked reader would be reading from.
|
||||
let (mut a, b) = UnixStream::pair().expect("socketpair");
|
||||
// The post-overflow state: the outbox is already closed.
|
||||
let mut outbox = Outbox::new();
|
||||
outbox.closed = true;
|
||||
let client = AttachClient {
|
||||
outbox: Arc::new((Mutex::new(outbox), Condvar::new())),
|
||||
shutdown_handle: b,
|
||||
frontend_id: FrontendId::LOCAL,
|
||||
server_protocol_version: PROTOCOL_VERSION,
|
||||
};
|
||||
// A send against the closed outbox fails *and* shuts the socket
|
||||
// down (F-008 fail-fast is now a real teardown, not just a flag).
|
||||
assert!(client.send_event(fe_key('x')).is_err());
|
||||
// The peer reads EOF: a real blocked reader would wake here and
|
||||
// fire Disconnected, instead of the session diverging silently.
|
||||
let mut buf = [0u8; 8];
|
||||
assert_eq!(
|
||||
a.read(&mut buf).expect("read peer"),
|
||||
0,
|
||||
"peer should see EOF after the shutdown"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -135,6 +135,30 @@ const MB_DROP_LINE_HEIGHT: f32 = 20.0;
|
|||
const MB_DROP_PAD_X: f32 = 10.0;
|
||||
const MB_DROP_MIN_WIDTH: f32 = 160.0;
|
||||
const MB_DROP_MAX_WIDTH: f32 = 480.0;
|
||||
|
||||
/// Visible slice of the completion dropdown given `n` shaped candidates,
|
||||
/// the `selected` index, and `band_top` pixels available above the status
|
||||
/// band (audit F-007). Returns `(first, count)` — `count` clamped to the
|
||||
/// rows that actually fit (so the box never renders above `y = 0`) and
|
||||
/// `first` scrolled to keep `selected` on screen. `None` when nothing can
|
||||
/// show: no candidates, or the window is too short for even one row. When
|
||||
/// the whole list fits this is `(0, n)`, identical to the pre-clamp
|
||||
/// behavior — the common path is unchanged.
|
||||
fn mb_dropdown_window(n: usize, selected: usize, band_top: f32) -> Option<(usize, usize)> {
|
||||
if n == 0 {
|
||||
return None;
|
||||
}
|
||||
let max_rows = (band_top / MB_DROP_ROW_HEIGHT).floor() as usize;
|
||||
if max_rows == 0 {
|
||||
return None;
|
||||
}
|
||||
let count = n.min(max_rows);
|
||||
let sel = selected.min(n - 1);
|
||||
// Anchor `sel` at the window's bottom edge when it would otherwise be
|
||||
// below the fold, then clamp so we never scroll past the last row.
|
||||
let first = sel.saturating_sub(count - 1).min(n - count);
|
||||
Some((first, count))
|
||||
}
|
||||
const QUAD_SHADER: &str = r"
|
||||
struct VertexOut {
|
||||
@builtin(position) pos: vec4<f32>,
|
||||
|
|
@ -751,7 +775,10 @@ impl ApplicationHandler<AppEvent> for App {
|
|||
Err(e) => {
|
||||
eprintln!("pmacs-gpu: attach failed: {e}");
|
||||
if let Some(state) = self.state.as_mut() {
|
||||
state.set_text("(attach failed; see stderr)");
|
||||
// Render a concise, actionable line in the window
|
||||
// itself (F-003) — e.g. a non-CRDT daemon — rather
|
||||
// than a generic "see stderr" the user won't read.
|
||||
state.set_text(&e.window_status());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -3235,17 +3262,29 @@ impl State {
|
|||
.shape_until_scroll(&mut self.font_system, false);
|
||||
}
|
||||
|
||||
/// The visible slice `(first, count)` of the dropdown candidates
|
||||
/// (audit F-007), clamped to the rows that fit above the band and
|
||||
/// scrolled to keep the selection on screen. `None` when closed,
|
||||
/// candidate-free, or too short for a row. See [`mb_dropdown_window`].
|
||||
fn mb_visible_window(&self) -> Option<(usize, usize)> {
|
||||
let mb = self.minibuffer.as_ref()?;
|
||||
let band_top = text_area_bottom(self.config.height);
|
||||
mb_dropdown_window(
|
||||
mb.candidates.len(),
|
||||
mb.selected.map_or(0, |s| s as usize),
|
||||
band_top,
|
||||
)
|
||||
}
|
||||
|
||||
/// Dropdown geometry `(left, top_y, width)` when the minibuffer has
|
||||
/// candidates: a list anchored just above the bottom band, growing
|
||||
/// upward, as wide as the widest candidate (clamped). `None` when
|
||||
/// closed or candidate-free. `refresh_mb_buffer` must have run so the
|
||||
/// width measurement is current.
|
||||
/// closed or candidate-free. The height is the *visible* row count
|
||||
/// (F-007), so `top_y` never goes above the window top.
|
||||
/// `refresh_mb_buffer` must have run so the width measurement is
|
||||
/// current.
|
||||
fn mb_dropdown_rect(&self) -> Option<(f32, f32, f32)> {
|
||||
let mb = self.minibuffer.as_ref()?;
|
||||
let n = mb.candidates.len();
|
||||
if n == 0 {
|
||||
return None;
|
||||
}
|
||||
let (_first, count) = self.mb_visible_window()?;
|
||||
let widest = self
|
||||
.mb_buffer
|
||||
.layout_runs()
|
||||
|
|
@ -3253,7 +3292,7 @@ impl State {
|
|||
.fold(0.0_f32, f32::max);
|
||||
let width = (widest + 2.0 * MB_DROP_PAD_X).clamp(MB_DROP_MIN_WIDTH, MB_DROP_MAX_WIDTH);
|
||||
let band_top = text_area_bottom(self.config.height);
|
||||
let top_y = band_top - n as f32 * MB_DROP_ROW_HEIGHT;
|
||||
let top_y = band_top - count as f32 * MB_DROP_ROW_HEIGHT;
|
||||
Some((STATUS_TEXT_PAD, top_y, width))
|
||||
}
|
||||
|
||||
|
|
@ -3263,21 +3302,28 @@ impl State {
|
|||
let Some(mb) = self.minibuffer.as_ref() else {
|
||||
return Vec::new();
|
||||
};
|
||||
let Some((first, count)) = self.mb_visible_window() else {
|
||||
return Vec::new();
|
||||
};
|
||||
let Some((x, top_y, width)) = self.mb_dropdown_rect() else {
|
||||
return Vec::new();
|
||||
};
|
||||
let n = mb.candidates.len();
|
||||
let mut rects = vec![MinimapRect {
|
||||
x,
|
||||
y: top_y,
|
||||
w: width,
|
||||
h: n as f32 * MB_DROP_ROW_HEIGHT,
|
||||
h: count as f32 * MB_DROP_ROW_HEIGHT,
|
||||
color: MENU_BG,
|
||||
}];
|
||||
if let Some(sel) = mb.selected {
|
||||
// Highlight the selection at its row *within the visible window*;
|
||||
// by construction it always falls inside [first, first + count).
|
||||
if let Some(sel) = mb.selected.map(|s| s as usize)
|
||||
&& sel >= first
|
||||
&& sel < first + count
|
||||
{
|
||||
rects.push(MinimapRect {
|
||||
x,
|
||||
y: top_y + sel as f32 * MB_DROP_ROW_HEIGHT,
|
||||
y: top_y + (sel - first) as f32 * MB_DROP_ROW_HEIGHT,
|
||||
w: width,
|
||||
h: MB_DROP_ROW_HEIGHT,
|
||||
color: MENU_SELECTED_BG,
|
||||
|
|
@ -3932,12 +3978,17 @@ impl State {
|
|||
.expect("menu text_renderer prepare");
|
||||
|
||||
// Q#MB1 — prepare the minibuffer dropdown glyphs in their layer.
|
||||
// The buffer is shaped with *all* candidates; F-007 scrolls it up
|
||||
// by `first` rows so line `first` lands at `top_y`, and the
|
||||
// existing `bounds.top`/`bottom` clip the rows scrolled out of the
|
||||
// visible window (no per-resize re-shape needed).
|
||||
let mb_areas: Vec<TextArea> = self
|
||||
.mb_dropdown_rect()
|
||||
.map(|(x, top_y, width)| TextArea {
|
||||
.mb_visible_window()
|
||||
.zip(self.mb_dropdown_rect())
|
||||
.map(|((first, _count), (x, top_y, width))| TextArea {
|
||||
buffer: &self.mb_buffer,
|
||||
left: x + MB_DROP_PAD_X,
|
||||
top: top_y,
|
||||
top: top_y - first as f32 * MB_DROP_ROW_HEIGHT,
|
||||
scale: 1.0,
|
||||
bounds: TextBounds {
|
||||
left: x as i32,
|
||||
|
|
@ -6103,6 +6154,32 @@ mod tests {
|
|||
assert!(!is_layout_text(Some("A"), Modifiers::SHIFT));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mb_dropdown_window_clamps_and_keeps_selection_visible() {
|
||||
// Row height is 20.0; a 1000px space fits any producer-capped list.
|
||||
assert_eq!(mb_dropdown_window(5, 2, 1000.0), Some((0, 5)));
|
||||
// The whole-fits path is identical to no clamp: (0, n).
|
||||
assert_eq!(mb_dropdown_window(10, 9, 1000.0), Some((0, 10)));
|
||||
|
||||
// A 100px space fits 5 rows. Selection near the top anchors at 0.
|
||||
assert_eq!(mb_dropdown_window(10, 0, 100.0), Some((0, 5)));
|
||||
// Selection past the fold scrolls so it stays visible (bottom edge).
|
||||
assert_eq!(mb_dropdown_window(10, 9, 100.0), Some((5, 5)));
|
||||
let (first, count) = mb_dropdown_window(10, 7, 100.0).unwrap();
|
||||
assert!(
|
||||
first <= 7 && 7 < first + count,
|
||||
"sel 7 in [{first},{})",
|
||||
first + count
|
||||
);
|
||||
|
||||
// Degenerate: too short for even one row ⇒ hide, never draw above 0.
|
||||
assert_eq!(mb_dropdown_window(10, 0, 10.0), None);
|
||||
// No candidates ⇒ nothing.
|
||||
assert_eq!(mb_dropdown_window(0, 0, 500.0), None);
|
||||
// An out-of-range selection is clamped, not panicked.
|
||||
assert_eq!(mb_dropdown_window(3, 99, 1000.0), Some((0, 3)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn translate_key_carries_modifiers() {
|
||||
use winit::keyboard::{Key as WKey, ModifiersState, NamedKey};
|
||||
|
|
|
|||
Loading…
Reference in New Issue