-
Notifications
You must be signed in to change notification settings - Fork 0
Workspaces 5/10: hold derived events in the transfer gap; drift cleanup #618
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
10d13e7
ebb6caf
dc4b8f6
29a1aba
969ea54
a323675
302b1f7
d0a80f4
2d2dc7a
8f6484f
32ee2c0
e0f73c2
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,21 @@ | ||
| // The sidecar's entry wires pty-core to the bundles it requires. These are | ||
| // source checks, because loading main.js needs the built bundles and a live | ||
| // stdin; each pins one injection whose absence fails silently at runtime. | ||
| const { test } = require('node:test'); | ||
| const assert = require('node:assert'); | ||
| const { readFileSync } = require('node:fs'); | ||
| const path = require('node:path'); | ||
|
|
||
| const source = readFileSync(path.join(__dirname, 'main.js'), 'utf8'); | ||
|
|
||
| test('pty-core is created with the shared sliceSince, so recovery capture reads a buffer', () => { | ||
| // Without it `outputSince` answers '' and `captureAgentRecovery` records | ||
| // nothing, with no error anywhere (pty-core.js -> outputSince). | ||
| assert.match(source, /\{\s*captureAgentRecovery,\s*createRecoveryStore,\s*sliceSince\s*\}\s*=\s*require\('\.\/recovery\.cjs'\)/); | ||
| assert.match(source, /nodePty,\s*\{\s*replay:\s*true,\s*sliceSince\s*\}\)/); | ||
| }); | ||
|
|
||
| test('recovery capture and the record take are answered from pty-core marks', () => { | ||
| assert.match(source, /receivedChars:\s*\(id\)\s*=>\s*mgr\.receivedChars\(id\)/); | ||
| assert.match(source, /outputSince:\s*\(id,\s*mark\)\s*=>\s*mgr\.outputSince\(id,\s*mark\)/); | ||
| }); |
| Original file line number | Diff line number | Diff line change | ||||||
|---|---|---|---|---|---|---|---|---|
|
|
@@ -86,6 +86,18 @@ struct RoutingState { | |||||||
| /// dor requestId -> the window handling it, so a cancel reaches the window | ||||||||
| /// holding the subscription, watch or completion claim it releases. | ||||||||
| dor_targets: HashMap<String, String>, | ||||||||
| /// Protocol events that arrived while their id was suppressed, delivered | ||||||||
| /// to the new owner behind its replay (`routing::Route::Hold`). Only ever | ||||||||
| /// emptied together with `awaiting_replay` (`lift_suppression`). | ||||||||
| held: HashMap<String, Vec<routing::HeldEvent>>, | ||||||||
| } | ||||||||
|
|
||||||||
| impl RoutingState { | ||||||||
| /// `routing::lift_suppression` over this state's two halves. The caller | ||||||||
| /// republishes `WindowState::suppressed` after it, still under the lock. | ||||||||
| fn lift_suppression(&mut self, id: &str) -> Vec<routing::HeldEvent> { | ||||||||
| routing::lift_suppression(&mut self.awaiting_replay, &mut self.held, id) | ||||||||
| } | ||||||||
| } | ||||||||
|
|
||||||||
| #[derive(Default)] | ||||||||
|
|
@@ -139,10 +151,10 @@ impl WindowState { | |||||||
| fn mint(&self, id: &str, label: &str) { | ||||||||
| let mut routing = guard(&self.routing); | ||||||||
| routing.owners.insert(id.to_string(), label.to_string()); | ||||||||
| if routing.awaiting_replay.remove(id).is_some() { | ||||||||
| self.suppressed | ||||||||
| .store(routing.awaiting_replay.len(), Ordering::Relaxed); | ||||||||
| } | ||||||||
| // Whatever was held belonged to the PTY that never arrived, not this one. | ||||||||
| routing.lift_suppression(id); | ||||||||
| self.suppressed | ||||||||
| .store(routing.awaiting_replay.len(), Ordering::Relaxed); | ||||||||
| } | ||||||||
|
|
||||||||
| /// Refuse every later `save_session` for `label` (a deliberate close removed | ||||||||
|
|
@@ -167,7 +179,12 @@ impl WindowState { | |||||||
| if suppress { | ||||||||
| routing.awaiting_replay.insert(id.clone(), now); | ||||||||
| } else { | ||||||||
| routing.awaiting_replay.remove(id); | ||||||||
| // A hand-back. The gap is lost here: the source was suppressed | ||||||||
| // like any other non-owner from the invoke on, and no replay | ||||||||
| // follows a hand-back, so the bytes and everything derived from | ||||||||
| // them are gone from its pane. A later stage recovers the gap | ||||||||
| // (docs/specs/standalone.md -> "Arrival queue"). | ||||||||
|
Comment on lines
+185
to
+186
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This points at a spec line that says the opposite. Either drop the promise here, or record it as a
Suggested change
|
||||||||
| routing.lift_suppression(id); | ||||||||
| } | ||||||||
| } | ||||||||
| self.suppressed | ||||||||
|
|
@@ -178,7 +195,7 @@ impl WindowState { | |||||||
| fn forget_pty(&self, id: &str) { | ||||||||
| let mut routing = guard(&self.routing); | ||||||||
| routing.owners.remove(id); | ||||||||
| routing.awaiting_replay.remove(id); | ||||||||
| routing.lift_suppression(id); | ||||||||
| self.suppressed | ||||||||
| .store(routing.awaiting_replay.len(), Ordering::Relaxed); | ||||||||
| } | ||||||||
|
|
@@ -190,7 +207,7 @@ impl WindowState { | |||||||
| fn clear_suppression(&self, ids: &[String]) { | ||||||||
| let mut routing = guard(&self.routing); | ||||||||
| for id in ids { | ||||||||
| routing.awaiting_replay.remove(id); | ||||||||
| routing.lift_suppression(id); | ||||||||
| } | ||||||||
| self.suppressed | ||||||||
| .store(routing.awaiting_replay.len(), Ordering::Relaxed); | ||||||||
|
|
@@ -209,7 +226,7 @@ impl WindowState { | |||||||
| let mut routing = guard(&self.routing); | ||||||||
| for id in lost.iter().flat_map(|arrival| &arrival.terminal_ids) { | ||||||||
| routing.owners.remove(id); | ||||||||
| routing.awaiting_replay.remove(id); | ||||||||
| routing.lift_suppression(id); | ||||||||
| } | ||||||||
| let owned = routing.owned_by(label); | ||||||||
| for id in &owned { | ||||||||
|
|
@@ -266,6 +283,8 @@ fn dispatch_sidecar_event(app: &AppHandle, event: &str, data: JsonValue) { | |||||||
| }; | ||||||||
|
|
||||||||
| let mut released: Vec<String> = Vec::new(); | ||||||||
| // Held events an expired suppression releases, flushed to the owner below. | ||||||||
| let mut flushed: Vec<(String, Vec<routing::HeldEvent>)> = Vec::new(); | ||||||||
| let delivery = { | ||||||||
| // Before the routing lock, never inside it (§`arrivals`). Nothing is | ||||||||
| // transferring in the steady state, so this second acquisition is paid | ||||||||
|
|
@@ -287,6 +306,13 @@ fn dispatch_sidecar_event(app: &AppHandle, event: &str, data: JsonValue) { | |||||||
| state | ||||||||
| .suppressed | ||||||||
| .store(routing.awaiting_replay.len(), Ordering::Relaxed); | ||||||||
| for id in &released { | ||||||||
| // The sweep already took the map entry; this takes the queue. | ||||||||
| let queue = routing.lift_suppression(id); | ||||||||
| if let (false, Some(label)) = (queue.is_empty(), routing.owners.get(id)) { | ||||||||
| flushed.push((label.clone(), queue)); | ||||||||
| } | ||||||||
| } | ||||||||
| } | ||||||||
| } | ||||||||
|
|
||||||||
|
|
@@ -300,6 +326,12 @@ fn dispatch_sidecar_event(app: &AppHandle, event: &str, data: JsonValue) { | |||||||
| }, | ||||||||
| ) { | ||||||||
| Route::Drop => Delivery::Nowhere, | ||||||||
| Route::Hold => { | ||||||||
| if let Some(id) = data.get("id").and_then(JsonValue::as_str) { | ||||||||
| routing::hold_event(&mut routing.held, id, event, data.clone()); | ||||||||
| } | ||||||||
| Delivery::Nowhere | ||||||||
| } | ||||||||
| Route::Broadcast => Delivery::Broadcast, | ||||||||
| Route::EmitTo(label) => Delivery::To(label.to_string()), | ||||||||
| // Resolved here, where the focus order is a sibling of the map the | ||||||||
|
|
@@ -318,6 +350,12 @@ fn dispatch_sidecar_event(app: &AppHandle, event: &str, data: JsonValue) { | |||||||
| } | ||||||||
| }; | ||||||||
|
|
||||||||
| for (label, queue) in flushed { | ||||||||
| for (held_event, held_data) in queue { | ||||||||
| let _ = app.emit_to(label.as_str(), held_event.as_str(), &held_data); | ||||||||
| } | ||||||||
| } | ||||||||
|
|
||||||||
| let mut delivered: Option<&str> = None; | ||||||||
| match &delivery { | ||||||||
| Delivery::Nowhere => {} | ||||||||
|
|
@@ -361,11 +399,21 @@ fn dispatch_sidecar_event(app: &AppHandle, event: &str, data: JsonValue) { | |||||||
| } | ||||||||
| "pty:replay" => { | ||||||||
| if let Some(id) = id() { | ||||||||
| let mut routing = guard(&state.routing); | ||||||||
| routing.awaiting_replay.remove(id); | ||||||||
| state | ||||||||
| .suppressed | ||||||||
| .store(routing.awaiting_replay.len(), Ordering::Relaxed); | ||||||||
| let queue = { | ||||||||
| let mut routing = guard(&state.routing); | ||||||||
| let queue = routing.lift_suppression(id); | ||||||||
| state | ||||||||
| .suppressed | ||||||||
| .store(routing.awaiting_replay.len(), Ordering::Relaxed); | ||||||||
| queue | ||||||||
| }; | ||||||||
| // Behind the replay, to the window that just received it: the | ||||||||
| // events describe bytes the replay carried. | ||||||||
| if let Some(label) = delivered { | ||||||||
| for (held_event, held_data) in queue { | ||||||||
| let _ = app.emit_to(label, held_event.as_str(), &held_data); | ||||||||
| } | ||||||||
| } | ||||||||
| } | ||||||||
| } | ||||||||
| "dor:controlRequest" => { | ||||||||
|
|
@@ -4334,12 +4382,56 @@ mod tests { | |||||||
| state.reassign(&["pane-a".to_string()], "ws-2", true); | ||||||||
| assert_eq!(state.suppressed.load(Ordering::Relaxed), 1); | ||||||||
|
|
||||||||
| super::routing::hold_event( | ||||||||
| &mut guard(&state.routing).held, | ||||||||
| "pane-a", | ||||||||
| "terminal:protocolEvents", | ||||||||
| serde_json::json!({"n": 1}), | ||||||||
| ); | ||||||||
|
|
||||||||
| state.mint("pane-a", "main"); | ||||||||
| assert!(guard(&state.routing).awaiting_replay.is_empty()); | ||||||||
| // Nothing held for the PTY that never arrived survives under its id. | ||||||||
| assert!(guard(&state.routing).held.is_empty()); | ||||||||
| assert_eq!(state.suppressed.load(Ordering::Relaxed), 0); | ||||||||
| assert_eq!(state.owned_by("main"), vec!["pane-a".to_string()]); | ||||||||
| } | ||||||||
|
|
||||||||
| /// Every way out of a suppression takes the held queue with it: a queue | ||||||||
| /// left behind would be flushed ahead of the *next* transfer's own gap. | ||||||||
| #[test] | ||||||||
| fn every_lift_of_a_suppression_takes_its_held_queue() { | ||||||||
| let state = super::WindowState::default(); | ||||||||
| let ids = ["pane-a".to_string()]; | ||||||||
| let queue_up = || { | ||||||||
| state.reassign(&ids, "ws-2", true); | ||||||||
| super::routing::hold_event( | ||||||||
| &mut guard(&state.routing).held, | ||||||||
| "pane-a", | ||||||||
| "terminal:protocolEvents", | ||||||||
| serde_json::json!({"n": 1}), | ||||||||
| ); | ||||||||
| assert_eq!(state.suppressed.load(Ordering::Relaxed), 1); | ||||||||
| }; | ||||||||
| let lifted = || { | ||||||||
| let routing = guard(&state.routing); | ||||||||
| routing.awaiting_replay.is_empty() && routing.held.is_empty() | ||||||||
| }; | ||||||||
|
|
||||||||
| queue_up(); | ||||||||
| state.clear_suppression(&ids); | ||||||||
| assert!(lifted()); | ||||||||
| assert_eq!(state.suppressed.load(Ordering::Relaxed), 0); | ||||||||
|
|
||||||||
| queue_up(); | ||||||||
| state.reassign(&ids, "main", false); | ||||||||
| assert!(lifted()); | ||||||||
|
|
||||||||
| queue_up(); | ||||||||
| state.forget_pty("pane-a"); | ||||||||
| assert!(lifted()); | ||||||||
| } | ||||||||
|
|
||||||||
| #[test] | ||||||||
| fn sweep_orphan_session_temps_removes_only_temps() { | ||||||||
| let dir = TempDir::new("sessions-sweep"); | ||||||||
|
|
||||||||
Uh oh!
There was an error while loading. Please reload this page.