| 1 | package main |
| 2 | |
| 3 | // Local takeover of a session held by a same-machine serve process — the |
| 4 | // desktop half of the single-writer handoff protocol (internal/serve's |
| 5 | // /ownership, /handoff, /external/frames, /reclaim and /mirror-end). |
| 6 | // |
| 7 | // The scenario: a remote desktop connected to this machine over SSH and left a |
| 8 | // resident serve holding session leases. The user now sits at THIS machine and |
| 9 | // opens the project locally. The tab lands in lease_blocked; instead of the |
| 10 | // dead-end banner it can now take the session over: Serve releases the lease, |
| 11 | // this window rebuilds on it, and the remote tab keeps watching through the |
| 12 | // frame mirror. When the remote side reclaims, this tab demotes itself to a |
| 13 | // read-only spectator fed by the same mirror in reverse. |
| 14 | |
| 15 | import ( |
| 16 | "bufio" |
| 17 | "context" |
| 18 | "encoding/json" |
| 19 | "fmt" |
| 20 | "io" |
| 21 | "log/slog" |
| 22 | "net/http" |
| 23 | "net/url" |
| 24 | "os" |
| 25 | "path/filepath" |
| 26 | "strings" |
| 27 | "sync" |
| 28 | "sync/atomic" |
| 29 | "time" |
| 30 | |
| 31 | "reasonix/internal/agent" |
| 32 | "reasonix/internal/config" |
| 33 | "reasonix/internal/event" |
| 34 | "reasonix/internal/eventwire" |
| 35 | "reasonix/internal/remote/bootstrap" |
| 36 | "reasonix/internal/store" |
| 37 | ) |
| 38 | |
| 39 | // SessionTakeoverView is what the confirmation dialog is built from. |
| 40 | type SessionTakeoverView struct { |
| 41 | Available bool `json:"available"` |
| 42 | Reason string `json:"reason,omitempty"` |
| 43 | SessionPath string `json:"sessionPath,omitempty"` |
| 44 | Holder string `json:"holder,omitempty"` // serve | external | other | free |
| 45 | RemoteAttached bool `json:"remoteAttached"` |
| 46 | Running bool `json:"running"` |
| 47 | Mirrored bool `json:"mirrored"` |
| 48 | HolderPID int `json:"holderPid,omitempty"` |
| 49 | HolderHost string `json:"holderHost,omitempty"` |
| 50 | } |
| 51 | |
| 52 | type takeoverGrant struct { |
| 53 | SessionPath string `json:"sessionPath"` |
| 54 | MirrorID string `json:"mirrorId"` |
| 55 | HandoffID string `json:"handoffId,omitempty"` |
| 56 | ReturnHandoffID string `json:"returnHandoffId"` |
| 57 | SourceWriterID string `json:"sourceWriterId"` |
| 58 | TargetWriterID string `json:"targetWriterId"` |
| 59 | Status string `json:"status"` |
| 60 | } |
| 61 | |
| 62 | // takeoverHandoffTimeout bounds the drain window for a wait-mode takeover. |
| 63 | const takeoverHandoffTimeout = 5 * time.Minute |
| 64 | |
| 65 | // takeoverAfterGrantHookForTest deterministically pauses a takeover after |
| 66 | // Serve published its grant but before the desktop enters the rebuild |
| 67 | // transaction. Production leaves it nil. |
| 68 | var takeoverAfterGrantHookForTest func() |
| 69 | |
| 70 | var takeoverFindTargetForTest func(context.Context, *App, string) (takeoverServeRecord, *http.Client, SessionTakeoverView, error) |
| 71 | |
| 72 | // takeoverServeRecord is one resident serve discovered from this machine's |
| 73 | // remote state directory, reachable over loopback HTTP. |
| 74 | type takeoverServeRecord struct { |
| 75 | slug string |
| 76 | state bootstrap.ServeState |
| 77 | base string |
| 78 | token string |
| 79 | } |
| 80 | |
| 81 | // discoverLocalTakeoverServes enumerates the serve state files under |
| 82 | // <Reasonix home>/remote. The bootstrap wrote them over SFTP; the takeover |
| 83 | // reads them locally because this machine is now where the user sits. |
| 84 | func discoverLocalTakeoverServes() []takeoverServeRecord { |
| 85 | dir := config.RemoteStateDir() |
| 86 | if dir == "" { |
| 87 | return nil |
| 88 | } |
| 89 | entries, err := os.ReadDir(dir) |
| 90 | if err != nil { |
| 91 | return nil |
| 92 | } |
| 93 | var out []takeoverServeRecord |
| 94 | for _, entry := range entries { |
| 95 | name := entry.Name() |
| 96 | if entry.IsDir() || !strings.HasPrefix(name, "serve-") || !strings.HasSuffix(name, ".json") { |
| 97 | continue |
| 98 | } |
| 99 | data, err := os.ReadFile(filepath.Join(dir, name)) |
| 100 | if err != nil { |
| 101 | continue |
| 102 | } |
| 103 | state, err := bootstrap.UnmarshalState(data) |
| 104 | if err != nil || state.PID <= 0 { |
| 105 | continue |
| 106 | } |
| 107 | if !takeoverProcessAlive(state.PID) { |
| 108 | // Stale state from a long-dead serve. Probing its dead port only |
| 109 | // burns handshakes (and can trip client/proxy rate limits). |
| 110 | continue |
| 111 | } |
| 112 | slug := strings.TrimSuffix(strings.TrimPrefix(name, "serve-"), ".json") |
| 113 | record := takeoverServeRecord{slug: slug, state: state} |
| 114 | // The port file carries the real bound address; the state JSON is the |
| 115 | // fallback for serves that predate it. |
| 116 | addr := state.Addr |
| 117 | if port, err := os.ReadFile(filepath.Join(dir, store.RemoteServePortName(slug))); err == nil { |
| 118 | if trimmed := strings.TrimSpace(string(port)); trimmed != "" { |
| 119 | addr = trimmed |
| 120 | } |
| 121 | } |
| 122 | if addr == "" { |
| 123 | continue |
| 124 | } |
| 125 | record.base = "http://" + addr |
| 126 | if token, err := os.ReadFile(filepath.Join(dir, store.RemoteServeTokenName(slug))); err == nil { |
| 127 | record.token = strings.TrimSpace(string(token)) |
| 128 | } |
| 129 | if record.token == "" { |
| 130 | continue |
| 131 | } |
| 132 | out = append(out, record) |
| 133 | } |
| 134 | return out |
| 135 | } |
| 136 | |
| 137 | var discoverLocalTakeoverServesForMirror = discoverLocalTakeoverServes |
| 138 | |
| 139 | // takeoverProcessAlive reports whether a pid is a live process on this host. |
| 140 | // The takeover protocol only ever talks to serves that are actually running; |
| 141 | // everything else is stale state. |
| 142 | func takeoverProcessAlive(pid int) bool { |
| 143 | return desktopProcessAlive(pid) |
| 144 | } |
| 145 | |
| 146 | func takeoverClient(ctx context.Context, record takeoverServeRecord) (*http.Client, error) { |
| 147 | client, err := newServeHTTPClient(record.base) |
| 148 | if err != nil { |
| 149 | return nil, err |
| 150 | } |
| 151 | if err := serveHandshake(ctx, client, record.base, record.token); err != nil { |
| 152 | return nil, err |
| 153 | } |
| 154 | return client, nil |
| 155 | } |
| 156 | |
| 157 | func takeoverOwnership(ctx context.Context, client *http.Client, base, sessionPath string) (SessionTakeoverView, error) { |
| 158 | query := url.Values{"session": []string{sessionPath}} |
| 159 | req, err := http.NewRequestWithContext(ctx, http.MethodGet, serveURL(base, "/ownership?"+query.Encode()), nil) |
| 160 | if err != nil { |
| 161 | return SessionTakeoverView{}, err |
| 162 | } |
| 163 | resp, err := client.Do(req) |
| 164 | if err != nil { |
| 165 | return SessionTakeoverView{}, err |
| 166 | } |
| 167 | defer resp.Body.Close() |
| 168 | body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) |
| 169 | if err != nil { |
| 170 | return SessionTakeoverView{}, err |
| 171 | } |
| 172 | if resp.StatusCode != http.StatusOK { |
| 173 | return SessionTakeoverView{}, fmt.Errorf("serve /ownership: status %d", resp.StatusCode) |
| 174 | } |
| 175 | var view SessionTakeoverView |
| 176 | if err := json.Unmarshal(body, &view); err != nil { |
| 177 | return SessionTakeoverView{}, err |
| 178 | } |
| 179 | return view, nil |
| 180 | } |
| 181 | |
| 182 | // findTakeoverTarget scans resident serves for one holding (or mirroring) the |
| 183 | // session, and returns a ready-to-use client for it. |
| 184 | func (a *App) findTakeoverTarget(ctx context.Context, sessionPath string) (takeoverServeRecord, *http.Client, SessionTakeoverView, error) { |
| 185 | records := discoverLocalTakeoverServesForMirror() |
| 186 | // Handshake failures back off: a serve whose token file was rotated by a |
| 187 | // later bootstrap would otherwise burn a failed login every probe. |
| 188 | records = a.serveProbesFresh(records) |
| 189 | var lastErr error |
| 190 | for _, record := range records { |
| 191 | client, err := takeoverClient(ctx, record) |
| 192 | if err != nil { |
| 193 | a.noteServeProbeFailure(record.base) |
| 194 | lastErr = err |
| 195 | continue |
| 196 | } |
| 197 | view, err := takeoverOwnership(ctx, client, record.base, sessionPath) |
| 198 | if err != nil { |
| 199 | lastErr = err |
| 200 | continue |
| 201 | } |
| 202 | if view.Holder == "serve" && !view.Mirrored { |
| 203 | return record, client, view, nil |
| 204 | } |
| 205 | } |
| 206 | if lastErr != nil { |
| 207 | return takeoverServeRecord{}, nil, SessionTakeoverView{}, fmt.Errorf("no reachable local serve holds this session: %w", lastErr) |
| 208 | } |
| 209 | return takeoverServeRecord{}, nil, SessionTakeoverView{}, fmt.Errorf("no resident serve on this machine holds this session") |
| 210 | } |
| 211 | |
| 212 | // QuerySessionTakeover reports whether the lease-blocked tab's session can be |
| 213 | // taken over from a local serve, plus the occupancy details the confirmation |
| 214 | // dialog shows (remote attached, turn running, holder identity). |
| 215 | func (a *App) QuerySessionTakeover(tabID string) (*SessionTakeoverView, error) { |
| 216 | tab := a.tabByID(tabID) |
| 217 | if tab == nil { |
| 218 | return nil, fmt.Errorf("unknown tab") |
| 219 | } |
| 220 | path := strings.TrimSpace(tab.currentSessionPath()) |
| 221 | if path == "" { |
| 222 | path = strings.TrimSpace(tab.SessionPath) |
| 223 | } |
| 224 | if path == "" { |
| 225 | return &SessionTakeoverView{Available: false, Reason: "tab has no session"}, nil |
| 226 | } |
| 227 | ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) |
| 228 | defer cancel() |
| 229 | _, _, view, err := a.findTakeoverTarget(ctx, path) |
| 230 | if err != nil { |
| 231 | return &SessionTakeoverView{Available: false, Reason: err.Error(), SessionPath: path}, nil |
| 232 | } |
| 233 | view.Available = true |
| 234 | view.SessionPath = path |
| 235 | return &view, nil |
| 236 | } |
| 237 | |
| 238 | // TakeoverSession performs the confirmed takeover: Serve releases the session |
| 239 | // (draining or cancelling its active turn per mode) and the tab's deferred |
| 240 | // rebuild picks the now-free lease up. mode is "wait" or "interrupt". |
| 241 | func (a *App) TakeoverSession(tabID, mode string) error { |
| 242 | tab := a.tabByID(tabID) |
| 243 | if tab == nil { |
| 244 | return fmt.Errorf("unknown tab") |
| 245 | } |
| 246 | if a.takeoverTabState(tab) == takeoverTabUnavailable { |
| 247 | return fmt.Errorf("tab is no longer waiting for a session lease") |
| 248 | } |
| 249 | path := strings.TrimSpace(tab.currentSessionPath()) |
| 250 | if path == "" { |
| 251 | path = strings.TrimSpace(tab.SessionPath) |
| 252 | } |
| 253 | if path == "" { |
| 254 | return fmt.Errorf("tab has no session") |
| 255 | } |
| 256 | if mode != "wait" && mode != "interrupt" { |
| 257 | mode = "wait" |
| 258 | } |
| 259 | a.mu.RLock() |
| 260 | sourceEpoch := a.runtimeEpochForTabLocked(tab) |
| 261 | a.mu.RUnlock() |
| 262 | ctx, cancel := context.WithTimeout(context.Background(), takeoverHandoffTimeout+30*time.Second) |
| 263 | defer cancel() |
| 264 | var record takeoverServeRecord |
| 265 | var client *http.Client |
| 266 | var view SessionTakeoverView |
| 267 | var err error |
| 268 | if takeoverFindTargetForTest != nil { |
| 269 | record, client, view, err = takeoverFindTargetForTest(ctx, a, path) |
| 270 | } else { |
| 271 | record, client, view, err = a.findTakeoverTarget(ctx, path) |
| 272 | } |
| 273 | if err != nil { |
| 274 | return err |
| 275 | } |
| 276 | if view.Mirrored || view.Holder != "serve" { |
| 277 | return fmt.Errorf("serve no longer holds this session (%s)", view.Holder) |
| 278 | } |
| 279 | body, err := json.Marshal(map[string]any{ |
| 280 | "sessionPath": path, |
| 281 | "targetWriterId": agent.SessionWriterID(), |
| 282 | "force": true, |
| 283 | "mode": mode, |
| 284 | "timeoutMs": takeoverHandoffTimeout.Milliseconds(), |
| 285 | }) |
| 286 | if err != nil { |
| 287 | return err |
| 288 | } |
| 289 | resp, err := serveDo(ctx, client, http.MethodPost, serveURL(record.base, "/handoff"), body) |
| 290 | if err != nil { |
| 291 | return fmt.Errorf("takeover handoff: %w", err) |
| 292 | } |
| 293 | defer resp.Body.Close() |
| 294 | respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<16)) |
| 295 | if resp.StatusCode != http.StatusOK { |
| 296 | return fmt.Errorf("takeover handoff: %s", strings.TrimSpace(string(respBody))) |
| 297 | } |
| 298 | var grant takeoverGrant |
| 299 | if err := json.Unmarshal(respBody, &grant); err != nil || grant.MirrorID == "" || grant.HandoffID == "" || grant.SourceWriterID == "" { |
| 300 | return fmt.Errorf("takeover handoff: invalid grant") |
| 301 | } |
| 302 | if grant.TargetWriterID != agent.SessionWriterID() { |
| 303 | a.endFailedTakeover(record, client, grant) |
| 304 | return fmt.Errorf("takeover handoff: grant targets another runtime") |
| 305 | } |
| 306 | if sessionRuntimeKey(grant.SessionPath) != "" && sessionRuntimeKey(grant.SessionPath) != sessionRuntimeKey(path) { |
| 307 | a.endFailedTakeover(record, client, grant) |
| 308 | return fmt.Errorf("takeover handoff: grant targets another session") |
| 309 | } |
| 310 | if hook := takeoverAfterGrantHookForTest; hook != nil { |
| 311 | hook() |
| 312 | } |
| 313 | |
| 314 | // The handoff request must not hold runtimeRebuildMu because Serve may wait |
| 315 | // for an active turn. Once the grant exists, however, validation, targeted |
| 316 | // acquisition, lease/mirror installation and controller publication are one |
| 317 | // serialized transaction. A concurrent tab rebuild therefore wins before |
| 318 | // acquisition or waits until this transaction commits; it can never be |
| 319 | // overwritten by a stale takeover. |
| 320 | a.runtimeRebuildMu.Lock() |
| 321 | defer a.runtimeRebuildMu.Unlock() |
| 322 | switch state := a.takeoverTabStateAt(tab, sourceEpoch, path); { |
| 323 | case state == takeoverTabUnavailable || tab.sessionLeaseRuntimeKey() != "": |
| 324 | a.endFailedTakeover(record, client, grant) |
| 325 | return fmt.Errorf("tab changed while taking over the session; retry") |
| 326 | case state == takeoverTabLocalSpectator: |
| 327 | return a.promoteLocalTakeoverSpectator(tab, path, sourceEpoch, record, client, grant) |
| 328 | } |
| 329 | lease, err := agent.TryAcquireSessionLeaseWithHandoff(path, grant.SourceWriterID, grant.HandoffID) |
| 330 | if err != nil { |
| 331 | a.endFailedTakeover(record, client, grant) |
| 332 | return userFacingSessionLeaseError("", err) |
| 333 | } |
| 334 | oldLease := tab.swapSessionLease(lease) |
| 335 | if oldLease != nil { |
| 336 | tab.swapSessionLease(oldLease) |
| 337 | if err := lease.ReleaseForHandoff(grant.SourceWriterID, grant.ReturnHandoffID); err != nil { |
| 338 | // This cannot be installed on the changed tab; let a return-only |
| 339 | // mirror retain and retry it without disturbing the old tab lease. |
| 340 | key := sessionRuntimeKey(path) |
| 341 | a.registerTakeoverMirror(key, tabID, path, record, client, grant) |
| 342 | a.takeoverMirrorForKey(key).holdPendingReturn(lease) |
| 343 | return fmt.Errorf("return takeover lease: %w", err) |
| 344 | } |
| 345 | a.endFailedTakeover(record, client, grant) |
| 346 | return fmt.Errorf("tab already owns another session lease") |
| 347 | } |
| 348 | |
| 349 | key := sessionRuntimeKey(path) |
| 350 | a.registerTakeoverMirror(key, tabID, path, record, client, grant) |
| 351 | previousStartup := tab.startupState() |
| 352 | pendingSequence := a.deferredRebuildSequence(tab.ID) |
| 353 | err = a.rebuildStartupTabLocked(tab) |
| 354 | if err == nil { |
| 355 | ctrl := a.controllerForTab(tab) |
| 356 | if ctrl == nil || sessionRuntimeKey(ctrl.SessionPath()) != key || tab.sessionLeaseRuntimeKey() != key { |
| 357 | err = fmt.Errorf("session startup did not publish the handed-off controller") |
| 358 | } |
| 359 | } |
| 360 | if err != nil { |
| 361 | if returned := tab.takeSessionLease(); returned != nil { |
| 362 | if m := a.takeoverMirrorForKey(key); m != nil { |
| 363 | m.returnLeaseAfterFailedTakeover(returned) |
| 364 | } else if releaseErr := returned.ReleaseForHandoff(grant.SourceWriterID, grant.ReturnHandoffID); releaseErr != nil { |
| 365 | tab.adoptSessionLease(returned) |
| 366 | } |
| 367 | } |
| 368 | a.mu.Lock() |
| 369 | if a.tabs[tab.ID] == tab && !tab.removed && tab.Ctrl == nil { |
| 370 | tab.restoreStartupState(previousStartup) |
| 371 | a.setSessionRuntimePhaseLocked(tab, sessionRuntimeLeaseBlocked, &sessionLeaseBusyError{}) |
| 372 | a.saveTabsLocked() |
| 373 | } |
| 374 | a.mu.Unlock() |
| 375 | return err |
| 376 | } |
| 377 | a.setTabReadOnly(tabID, false) |
| 378 | a.clearDeferredRebuildVersion(tabID, pendingSequence) |
| 379 | return nil |
| 380 | } |
| 381 | |
| 382 | func (a *App) endFailedTakeover(record takeoverServeRecord, client *http.Client, grant takeoverGrant) { |
| 383 | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 384 | defer cancel() |
| 385 | payload, _ := json.Marshal(map[string]string{"sessionPath": grant.SessionPath, "mirrorId": grant.MirrorID}) |
| 386 | resp, err := serveDo(ctx, client, http.MethodPost, serveURL(record.base, "/mirror-end"), payload) |
| 387 | if err == nil { |
| 388 | _, _ = io.Copy(io.Discard, resp.Body) |
| 389 | resp.Body.Close() |
| 390 | } |
| 391 | } |
| 392 | |
| 393 | // takeoverMirror forwards one local tab's events to the serve that used to own |
| 394 | // the session, so the remote tab keeps rendering, and watches the heartbeat |
| 395 | // responses for a reclaim request. One mirror per session key. |
| 396 | type takeoverMirror struct { |
| 397 | app *App |
| 398 | key string |
| 399 | sessionPath string |
| 400 | |
| 401 | // sendMu serializes every binding user and writer: HTTP forwarding, |
| 402 | // re-adoption, grant replacement, reverse reservation, and mirror-end. |
| 403 | // Code holding sendMu may take mu; the inverse order is forbidden. |
| 404 | sendMu sync.Mutex |
| 405 | mu sync.Mutex |
| 406 | tabID string |
| 407 | sink *tabEventSink |
| 408 | client *http.Client |
| 409 | record takeoverServeRecord |
| 410 | grant takeoverGrant |
| 411 | bindingRevision uint64 |
| 412 | queue eventwire.MirrorQueue |
| 413 | pendingReturn *agent.SessionLease |
| 414 | returnNextTry time.Time |
| 415 | returnBackoff time.Duration |
| 416 | releaseHandoff func(*agent.SessionLease, string, string) error |
| 417 | |
| 418 | reclaimRequested atomic.Bool |
| 419 | returned atomic.Bool |
| 420 | stopping atomic.Bool |
| 421 | // closing marks a tab close that owns the farewell: it releases the writer |
| 422 | // first, so Serve hands the session back without waiting for it to drop. |
| 423 | closing atomic.Bool |
| 424 | // ended records a delivered farewell so the close epilogue and the loop |
| 425 | // cannot send it twice; a failed send stays unset for the retry. |
| 426 | ended atomic.Bool |
| 427 | consecutiveFailures int32 |
| 428 | stop chan struct{} |
| 429 | done chan struct{} |
| 430 | wake chan struct{} |
| 431 | stopOnce sync.Once |
| 432 | detachOnce sync.Once |
| 433 | } |
| 434 | |
| 435 | const ( |
| 436 | takeoverMirrorMaxQueue = eventwire.MirrorBatchMaxFrames |
| 437 | takeoverMirrorFlushEvery = 120 * time.Millisecond |
| 438 | takeoverMirrorHeartbeat = 5 * time.Second |
| 439 | ) |
| 440 | |
| 441 | // adoptSessionFromLocalServe announces a directly-opened local session to the |
| 442 | // resident serve on this machine. Without a handoff there is no mirrored |
| 443 | // entry, so the remote side could neither watch it live nor reclaim it — it |
| 444 | // only saw the raw 409 lease refusal from serve's /resume. Adoption registers |
| 445 | // the mirror (and its frame forwarder), which restores both: the remote tab |
| 446 | // can spectator-attach, and its take-back works through /reclaim. |
| 447 | // |
| 448 | // Skipped when the serve itself holds the session (the /handoff takeover |
| 449 | // flow governs), when the serve is unreachable, or when the session is |
| 450 | // already mirrored by another writer. |
| 451 | |
| 452 | // serveProbeBackoffWindow suppresses probing a serve whose handshake failed |
| 453 | // recently - typically a token file rotated by a later bootstrap - so a |
| 454 | // polling loop cannot hammer it with failed logins. |
| 455 | const serveProbeBackoffWindow = 60 * time.Second |
| 456 | |
| 457 | func (a *App) serveProbesFresh(records []takeoverServeRecord) []takeoverServeRecord { |
| 458 | a.serveProbeMu.Lock() |
| 459 | defer a.serveProbeMu.Unlock() |
| 460 | if a.serveProbeUntil == nil { |
| 461 | return records |
| 462 | } |
| 463 | now := time.Now() |
| 464 | out := records[:0] |
| 465 | for _, record := range records { |
| 466 | if until := a.serveProbeUntil[record.base]; until.After(now) { |
| 467 | continue |
| 468 | } |
| 469 | out = append(out, record) |
| 470 | } |
| 471 | return out |
| 472 | } |
| 473 | |
| 474 | func (a *App) noteServeProbeFailure(base string) { |
| 475 | a.serveProbeMu.Lock() |
| 476 | if a.serveProbeUntil == nil { |
| 477 | a.serveProbeUntil = map[string]time.Time{} |
| 478 | } |
| 479 | a.serveProbeUntil[base] = time.Now().Add(serveProbeBackoffWindow) |
| 480 | a.serveProbeMu.Unlock() |
| 481 | } |
| 482 | |
| 483 | // pathWithinDir reports whether child is inside dir (canonical path prefix). |
| 484 | func pathWithinDir(child, dir string) bool { |
| 485 | child = strings.TrimRight(filepath.Clean(child), string(filepath.Separator)) + string(filepath.Separator) |
| 486 | dir = strings.TrimRight(filepath.Clean(dir), string(filepath.Separator)) + string(filepath.Separator) |
| 487 | return strings.HasPrefix(strings.ToLower(child), strings.ToLower(dir)) |
| 488 | } |
| 489 | |
| 490 | func (a *App) registerTakeoverMirror(key, tabID, sessionPath string, record takeoverServeRecord, client *http.Client, grant takeoverGrant) { |
| 491 | if key == "" { |
| 492 | return |
| 493 | } |
| 494 | var m *takeoverMirror |
| 495 | for { |
| 496 | a.takeoverMu.Lock() |
| 497 | if a.takeoverMirrors == nil { |
| 498 | a.takeoverMirrors = map[string]*takeoverMirror{} |
| 499 | } |
| 500 | m = a.takeoverMirrors[key] |
| 501 | if m == nil || m.returned.Load() || m.stopping.Load() { |
| 502 | m = newTakeoverMirror(a, key, tabID, sessionPath, nil, record, client, grant) |
| 503 | a.takeoverMirrors[key] = m |
| 504 | a.takeoverMu.Unlock() |
| 505 | go m.run(client, record) |
| 506 | break |
| 507 | } |
| 508 | a.takeoverMu.Unlock() |
| 509 | m.sendMu.Lock() |
| 510 | a.takeoverMu.Lock() |
| 511 | current := a.takeoverMirrors[key] == m |
| 512 | a.takeoverMu.Unlock() |
| 513 | if !current || m.returned.Load() || m.stopping.Load() { |
| 514 | m.sendMu.Unlock() |
| 515 | continue |
| 516 | } |
| 517 | m.mu.Lock() |
| 518 | m.tabID = tabID |
| 519 | m.record = record |
| 520 | m.client = client |
| 521 | m.grant = grant |
| 522 | m.bindingRevision++ |
| 523 | m.consecutiveFailures = 0 |
| 524 | m.mu.Unlock() |
| 525 | m.sendMu.Unlock() |
| 526 | break |
| 527 | } |
| 528 | a.attachTakeoverMirror(tabID, sessionPath) |
| 529 | } |
| 530 | |
| 531 | // attachTakeoverMirror points the tab's current sink at its mirror, if one is |
| 532 | // registered for the session. Called after every successful (re)bind so a |
| 533 | // deferred rebuild or a later session switch keeps the wiring true. |
| 534 | func (a *App) attachTakeoverMirror(tabID, sessionPath string) { |
| 535 | key := sessionRuntimeKey(sessionPath) |
| 536 | if key == "" { |
| 537 | return |
| 538 | } |
| 539 | a.takeoverMu.Lock() |
| 540 | m := a.takeoverMirrors[key] |
| 541 | a.takeoverMu.Unlock() |
| 542 | if m == nil { |
| 543 | return |
| 544 | } |
| 545 | a.mu.RLock() |
| 546 | tab := a.tabByIDLocked(tabID) |
| 547 | var sink *tabEventSink |
| 548 | if tab != nil { |
| 549 | sink = tab.sink |
| 550 | } |
| 551 | a.mu.RUnlock() |
| 552 | if tab == nil || sink == nil { |
| 553 | return |
| 554 | } |
| 555 | m.mu.Lock() |
| 556 | m.tabID = tabID |
| 557 | m.sink = sink |
| 558 | m.mu.Unlock() |
| 559 | sink.setTakeoverMirror(m) |
| 560 | } |
| 561 | |
| 562 | func (a *App) takeoverMirrorForKey(key string) *takeoverMirror { |
| 563 | if key == "" { |
| 564 | return nil |
| 565 | } |
| 566 | a.takeoverMu.Lock() |
| 567 | defer a.takeoverMu.Unlock() |
| 568 | return a.takeoverMirrors[key] |
| 569 | } |
| 570 | |
| 571 | // stopTakeoverMirrors halts every mirror's forwarding loop. The registry |
| 572 | // entries stay: endTakeoverMirrors still needs them after controller teardown |
| 573 | // has released the leases to tell Serve the writers are gone. |
| 574 | func (a *App) stopTakeoverMirrors() { |
| 575 | for _, m := range a.snapshotTakeoverMirrors() { |
| 576 | m.stopLoop() |
| 577 | } |
| 578 | } |
| 579 | |
| 580 | // endTakeoverMirrors is the shutdown epilogue: every lease is released by now, |
| 581 | // so Serve's mirror-end immediately hands the sessions back to their remote |
| 582 | // tabs instead of waiting out the stale-mirror timeout. |
| 583 | func (a *App) endTakeoverMirrors() { |
| 584 | for _, m := range a.snapshotTakeoverMirrors() { |
| 585 | if m.finalizePendingReturn() { |
| 586 | m.mirrorEnd() |
| 587 | } |
| 588 | m.detach() |
| 589 | } |
| 590 | } |
| 591 | |
| 592 | func (a *App) snapshotTakeoverMirrors() []*takeoverMirror { |
| 593 | a.takeoverMu.Lock() |
| 594 | mirrors := make([]*takeoverMirror, 0, len(a.takeoverMirrors)) |
| 595 | for _, m := range a.takeoverMirrors { |
| 596 | mirrors = append(mirrors, m) |
| 597 | } |
| 598 | a.takeoverMu.Unlock() |
| 599 | return mirrors |
| 600 | } |
| 601 | |
| 602 | // returnTakeoverLeaseForShutdown publishes the reverse reservation while the |
| 603 | // tab still owns its lease. The controller snapshot has already completed in |
| 604 | // shutdownBody; a failed reservation leaves the lease installed so the normal |
| 605 | // release fallback can still make progress. |
| 606 | func (a *App) returnTakeoverLeaseForShutdown(tab *WorkspaceTab) bool { |
| 607 | if tab == nil { |
| 608 | return false |
| 609 | } |
| 610 | path := strings.TrimSpace(tab.currentSessionPath()) |
| 611 | m := a.takeoverMirrorForKey(sessionRuntimeKey(path)) |
| 612 | if m == nil { |
| 613 | return false |
| 614 | } |
| 615 | _, _, _, grant := m.snapshotClient() |
| 616 | if grant.SourceWriterID == "" || grant.ReturnHandoffID == "" { |
| 617 | return false |
| 618 | } |
| 619 | lease := tab.takeSessionLease() |
| 620 | if lease == nil { |
| 621 | return false |
| 622 | } |
| 623 | if err := lease.ReleaseForHandoff(grant.SourceWriterID, grant.ReturnHandoffID); err != nil { |
| 624 | tab.adoptSessionLease(lease) |
| 625 | slog.Warn("desktop: reserve takeover lease return during shutdown", "session", path, "err", err) |
| 626 | return false |
| 627 | } |
| 628 | return true |
| 629 | } |
| 630 | |
| 631 | // forwardEvent enqueues one local event for the mirror. Called from the tab |
| 632 | // sink's Emit; must never block the agent loop. |
| 633 | func (m *takeoverMirror) forwardEvent(e event.Event) { |
| 634 | if m == nil { |
| 635 | return |
| 636 | } |
| 637 | wired := eventwire.ToWire(e) |
| 638 | m.mu.Lock() |
| 639 | m.queue.Push(wired) |
| 640 | m.mu.Unlock() |
| 641 | select { |
| 642 | case m.wake <- struct{}{}: |
| 643 | default: |
| 644 | } |
| 645 | } |
| 646 | |
| 647 | // holdPendingReturn transfers a failed takeover target into the mirror. The |
| 648 | // tab's prior state stays intact while this independent lease fences out third |
| 649 | // writers until its reverse reservation is durably published. |
| 650 | func (m *takeoverMirror) holdPendingReturn(lease *agent.SessionLease) { |
| 651 | if m == nil || lease == nil { |
| 652 | return |
| 653 | } |
| 654 | m.mu.Lock() |
| 655 | if m.pendingReturn != nil && m.pendingReturn != lease { |
| 656 | m.mu.Unlock() |
| 657 | lease.Release() |
| 658 | return |
| 659 | } |
| 660 | m.pendingReturn = lease |
| 661 | m.returnBackoff = 200 * time.Millisecond |
| 662 | m.returnNextTry = time.Now() |
| 663 | m.mu.Unlock() |
| 664 | select { |
| 665 | case m.wake <- struct{}{}: |
| 666 | default: |
| 667 | } |
| 668 | } |
| 669 | |
| 670 | func (m *takeoverMirror) returnLeaseAfterFailedTakeover(lease *agent.SessionLease) { |
| 671 | m.holdPendingReturn(lease) |
| 672 | } |
| 673 | |
| 674 | // retryPendingReturn reports true only when it published a pending reverse |
| 675 | // reservation. The run loop then ends the mirror from its own goroutine, |
| 676 | // avoiding stop/join self-deadlock. |
| 677 | func (m *takeoverMirror) retryPendingReturn(force bool) bool { |
| 678 | if m == nil { |
| 679 | return false |
| 680 | } |
| 681 | m.sendMu.Lock() |
| 682 | defer m.sendMu.Unlock() |
| 683 | m.mu.Lock() |
| 684 | lease := m.pendingReturn |
| 685 | grant := m.grant |
| 686 | nextTry := m.returnNextTry |
| 687 | release := m.releaseHandoff |
| 688 | m.mu.Unlock() |
| 689 | if lease == nil || (!force && time.Now().Before(nextTry)) { |
| 690 | return false |
| 691 | } |
| 692 | var err error |
| 693 | if release != nil { |
| 694 | err = release(lease, grant.SourceWriterID, grant.ReturnHandoffID) |
| 695 | } else { |
| 696 | err = lease.ReleaseForHandoff(grant.SourceWriterID, grant.ReturnHandoffID) |
| 697 | } |
| 698 | m.mu.Lock() |
| 699 | defer m.mu.Unlock() |
| 700 | if m.pendingReturn != lease { |
| 701 | return false |
| 702 | } |
| 703 | if err == nil { |
| 704 | m.pendingReturn = nil |
| 705 | m.returnBackoff = 0 |
| 706 | m.returnNextTry = time.Time{} |
| 707 | // Fence grant replacement before the run loop sends mirror-end. A new |
| 708 | // registration will create its own mirror instead of rotating the |
| 709 | // generation whose target lease was just returned. |
| 710 | m.returned.Store(true) |
| 711 | return true |
| 712 | } |
| 713 | if m.returnBackoff <= 0 { |
| 714 | m.returnBackoff = 200 * time.Millisecond |
| 715 | } else { |
| 716 | m.returnBackoff = min(m.returnBackoff*2, 5*time.Second) |
| 717 | } |
| 718 | m.returnNextTry = time.Now().Add(m.returnBackoff) |
| 719 | return false |
| 720 | } |
| 721 | |
| 722 | // finalizePendingReturn runs at final app teardown. A reservation that still |
| 723 | // cannot be written keeps its lease until this point, then releases the OS |
| 724 | // lock without sending mirror-end; Serve will recover it as a vanished writer. |
| 725 | func (m *takeoverMirror) finalizePendingReturn() bool { |
| 726 | if m == nil { |
| 727 | return true |
| 728 | } |
| 729 | if m.retryPendingReturn(true) { |
| 730 | return true |
| 731 | } |
| 732 | m.mu.Lock() |
| 733 | lease := m.pendingReturn |
| 734 | m.pendingReturn = nil |
| 735 | m.mu.Unlock() |
| 736 | if lease != nil { |
| 737 | lease.Release() |
| 738 | return false |
| 739 | } |
| 740 | return true |
| 741 | } |
| 742 | |
| 743 | func (m *takeoverMirror) snapshotClient() (*http.Client, takeoverServeRecord, string, takeoverGrant) { |
| 744 | m.mu.Lock() |
| 745 | defer m.mu.Unlock() |
| 746 | return m.client, m.record, m.tabID, m.grant |
| 747 | } |
| 748 | |
| 749 | func (m *takeoverMirror) snapshotBinding() (*http.Client, takeoverServeRecord, string, takeoverGrant, uint64) { |
| 750 | m.mu.Lock() |
| 751 | defer m.mu.Unlock() |
| 752 | return m.client, m.record, m.tabID, m.grant, m.bindingRevision |
| 753 | } |
| 754 | |
| 755 | func (m *takeoverMirror) bindingCurrent(client *http.Client, grant takeoverGrant, revision uint64) bool { |
| 756 | m.mu.Lock() |
| 757 | defer m.mu.Unlock() |
| 758 | return m.client == client && m.grant.MirrorID == grant.MirrorID && m.bindingRevision == revision |
| 759 | } |
| 760 | |
| 761 | // takeoverTabLive reports whether the mirrored session still has a live |
| 762 | // desktop runtime (visible tab or detached background runtime). |
| 763 | func (a *App) takeoverTabLive(sessionPath string) bool { |
| 764 | return a.sessionParentLive(sessionPath) |
| 765 | } |
| 766 | |
| 767 | // tabHoldingSession returns the live tab currently owning the session path, |
| 768 | // using the same liveness notion as sessionParentLive. Mirrors outlive tab |
| 769 | // rebuilds, so lease-affecting actions must resolve the tab through the |
| 770 | // session rather than a snapshotted tab ID. |
| 771 | func (a *App) tabHoldingSession(sessionPath string) *WorkspaceTab { |
| 772 | key := sessionRuntimeKey(sessionPath) |
| 773 | if a == nil || key == "" { |
| 774 | return nil |
| 775 | } |
| 776 | a.mu.RLock() |
| 777 | defer a.mu.RUnlock() |
| 778 | holding := func(tab *WorkspaceTab) bool { |
| 779 | if tab == nil { |
| 780 | return false |
| 781 | } |
| 782 | if sessionRuntimeKey(tab.SessionPath) == key { |
| 783 | return true |
| 784 | } |
| 785 | return tab.Ctrl != nil && sessionRuntimeKey(tab.Ctrl.SessionPath()) == key |
| 786 | } |
| 787 | for _, tab := range a.tabs { |
| 788 | if holding(tab) { |
| 789 | return tab |
| 790 | } |
| 791 | } |
| 792 | for _, tab := range a.detachedSessions { |
| 793 | if holding(tab) { |
| 794 | return tab |
| 795 | } |
| 796 | } |
| 797 | return nil |
| 798 | } |
| 799 | |
| 800 | // requestDemote reacts to a remote reclaim: this tab loses speaking rights. |
| 801 | // The demotion itself is passive — flip the tab read-only, release the lease, |
| 802 | // tell the user, and let Serve resume ownership. |
| 803 | func (m *takeoverMirror) requestDemote(mode string) { |
| 804 | if !m.reclaimRequested.CompareAndSwap(false, true) { |
| 805 | return |
| 806 | } |
| 807 | m.app.emitTakeoverNotice(m, event.LevelWarn, "session_reclaim_requested", |
| 808 | "The remote side is taking this session back; this window is now read-only.") |
| 809 | go m.demote(mode == string(handoffModeInterruptLocal)) |
| 810 | } |
| 811 | |
| 812 | const handoffModeInterruptLocal = "interrupt" |
| 813 | |
| 814 | func (m *takeoverMirror) demote(interrupt bool) { |
| 815 | a := m.app |
| 816 | tab := a.tabByID(m.tabIDSnapshot()) |
| 817 | if tab == nil { |
| 818 | // Tab IDs rotate on rebuilds and app restarts while a reclaim is in |
| 819 | // flight; the session path is the stable handle. Without this |
| 820 | // fallback the demote skipped the read-only flip and the lease |
| 821 | // release, and the remote reclaim waited on a writer that had |
| 822 | // already forgotten it was one. |
| 823 | tab = a.tabHoldingSession(m.sessionPath) |
| 824 | } |
| 825 | var sink *tabEventSink |
| 826 | if tab != nil { |
| 827 | a.mu.RLock() |
| 828 | sink = tab.sink |
| 829 | a.mu.RUnlock() |
| 830 | // Block new submits before waiting for the current turn to drain. Keep |
| 831 | // integrated terminals alive: reclaim transfers this session's writer, |
| 832 | // not ownership of the local terminal process or workspace. |
| 833 | a.setTabReadOnlyPreservingTerminals(tab.ID, true) |
| 834 | } |
| 835 | m.emitNoticeSink(sink, event.LevelWarn, "session_taken_over_local", |
| 836 | "This session was taken back by the remote side. This window is a read-only spectator.") |
| 837 | if interrupt && tab != nil && tab.Ctrl != nil { |
| 838 | // The remote asked to cancel an in-flight local turn; with the tab |
| 839 | // read-only the admission gate refuses new input, cancel what is |
| 840 | // already running so the drain completes quickly. |
| 841 | tab.Ctrl.Cancel() |
| 842 | } |
| 843 | if tab == nil || tab.Ctrl == nil { |
| 844 | return |
| 845 | } |
| 846 | deadline := time.Now().Add(takeoverHandoffTimeout) |
| 847 | for controllerHasActiveRuntimeWork(tab.Ctrl) && time.Now().Before(deadline) { |
| 848 | time.Sleep(50 * time.Millisecond) |
| 849 | } |
| 850 | if controllerHasActiveRuntimeWork(tab.Ctrl) { |
| 851 | a.setTabReadOnly(tab.ID, false) |
| 852 | return |
| 853 | } |
| 854 | if err := tab.Ctrl.Snapshot(); err != nil { |
| 855 | slog.Warn("desktop: snapshot before returning takeover", "session", m.sessionPath, "err", err) |
| 856 | a.setTabReadOnly(tab.ID, false) |
| 857 | return |
| 858 | } |
| 859 | if err := m.returnLeaseForDemotion(tab); err != nil { |
| 860 | a.setTabReadOnly(tab.ID, false) |
| 861 | return |
| 862 | } |
| 863 | a.markLocalTakeoverSpectator(tab) |
| 864 | m.stopAndFinalize(false) |
| 865 | m.startSpectate(tab, sink) |
| 866 | } |
| 867 | |
| 868 | // returnLeaseForDemotion waits for any in-flight sender/re-adoption, then uses |
| 869 | // one stable binding for the reverse reservation and mirror-end. returned is |
| 870 | // published before releasing sendMu, so the forwarding loop cannot issue a |
| 871 | // later request from the retired generation while demote stops and joins it. |
| 872 | func (m *takeoverMirror) returnLeaseForDemotion(tab *WorkspaceTab) error { |
| 873 | if m == nil || tab == nil { |
| 874 | return fmt.Errorf("takeover return target unavailable") |
| 875 | } |
| 876 | m.sendMu.Lock() |
| 877 | defer m.sendMu.Unlock() |
| 878 | client, record, _, grant := m.snapshotClient() |
| 879 | if client == nil || grant.MirrorID == "" || grant.SourceWriterID == "" || grant.ReturnHandoffID == "" { |
| 880 | return fmt.Errorf("takeover return binding unavailable") |
| 881 | } |
| 882 | lease := tab.takeSessionLease() |
| 883 | if lease == nil { |
| 884 | return fmt.Errorf("takeover session lease unavailable") |
| 885 | } |
| 886 | if err := lease.ReleaseForHandoff(grant.SourceWriterID, grant.ReturnHandoffID); err != nil { |
| 887 | tab.adoptSessionLease(lease) |
| 888 | return err |
| 889 | } |
| 890 | m.returned.Store(true) |
| 891 | m.mirrorEndLocked(client, record, grant) |
| 892 | return nil |
| 893 | } |
| 894 | |
| 895 | // emitTakeoverNotice surfaces a takeover lifecycle change as a notice frame on |
| 896 | // the tab's event channel (best effort; the sink may not exist yet). |
| 897 | func (a *App) emitTakeoverNotice(m *takeoverMirror, level event.Level, code, text string) { |
| 898 | m.mu.Lock() |
| 899 | sink := m.sink |
| 900 | m.mu.Unlock() |
| 901 | m.emitNoticeSink(sink, level, code, text) |
| 902 | } |
| 903 | |
| 904 | func (m *takeoverMirror) emitNoticeSink(sink *tabEventSink, level event.Level, code, text string) { |
| 905 | if sink == nil { |
| 906 | return |
| 907 | } |
| 908 | tabID, _ := sink.binding() |
| 909 | e := event.Event{Kind: event.Notice, Level: level, Code: code, Text: text, SessionPath: m.sessionPath} |
| 910 | sink.emitRuntimeEvent(eventChannel, toWireTabWithSubmission(e, tabID, sink.runtimeEpochSnapshot(), "", 0, sink.sessionGenerationSnapshot())) |
| 911 | } |
| 912 | |
| 913 | // mirrorEnd tells Serve the writer is gone so the remote side resumes without |
| 914 | // waiting for the stale-mirror timeout. Best effort. |
| 915 | func (m *takeoverMirror) mirrorEnd() { |
| 916 | m.sendMu.Lock() |
| 917 | defer m.sendMu.Unlock() |
| 918 | m.mu.Lock() |
| 919 | pending := m.pendingReturn != nil |
| 920 | m.mu.Unlock() |
| 921 | if pending { |
| 922 | return |
| 923 | } |
| 924 | client, record, _, grant := m.snapshotClient() |
| 925 | if client == nil || grant.MirrorID == "" { |
| 926 | return |
| 927 | } |
| 928 | m.mirrorEndLocked(client, record, grant) |
| 929 | } |
| 930 | |
| 931 | func (m *takeoverMirror) mirrorEndLocked(client *http.Client, record takeoverServeRecord, grant takeoverGrant) { |
| 932 | if m.ended.Load() { |
| 933 | return |
| 934 | } |
| 935 | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 936 | defer cancel() |
| 937 | payload, _ := json.Marshal(map[string]string{"sessionPath": m.sessionPath, "mirrorId": grant.MirrorID}) |
| 938 | resp, err := serveDo(ctx, client, http.MethodPost, serveURL(record.base, "/mirror-end"), payload) |
| 939 | if err != nil { |
| 940 | // Undelivered: leave the farewell unmarked so a later path retries. |
| 941 | return |
| 942 | } |
| 943 | _, _ = io.Copy(io.Discard, resp.Body) |
| 944 | resp.Body.Close() |
| 945 | m.ended.Store(true) |
| 946 | } |
| 947 | |
| 948 | // claimTakeoverMirrorFarewell muzzles the mirror loop's farewell while the |
| 949 | // closing tab still owns its writer. The returned release runs once the close |
| 950 | // has finished with that writer — released, or retained by an abandoned close |
| 951 | // — so the loop is only ever unmuzzled when announcing is safe. |
| 952 | func (a *App) claimTakeoverMirrorFarewell(sessionPath string) (*takeoverMirror, func()) { |
| 953 | m := a.takeoverMirrorForKey(sessionRuntimeKey(sessionPath)) |
| 954 | if m == nil { |
| 955 | return nil, func() {} |
| 956 | } |
| 957 | m.closing.Store(true) |
| 958 | return m, func() { m.closing.Store(false) } |
| 959 | } |
| 960 | |
| 961 | // endTakeoverMirrorForClosedTab is the per-tab analogue of endTakeoverMirrors: |
| 962 | // the writer is released, so Serve can hand the session back at once. |
| 963 | func (a *App) endTakeoverMirrorForClosedTab(m *takeoverMirror) { |
| 964 | if m == nil { |
| 965 | return |
| 966 | } |
| 967 | // Off the close path's locks: a bounded HTTP round trip must not hold the |
| 968 | // runtime mutation barrier. |
| 969 | a.goSafe("endTakeoverMirrorForClosedTab", func() { |
| 970 | if m.finalizePendingReturn() { |
| 971 | m.mirrorEnd() |
| 972 | } |
| 973 | m.detach() |
| 974 | }) |
| 975 | } |
| 976 | |
| 977 | // stopLoop halts the forwarding goroutine. Idempotent. |
| 978 | func (m *takeoverMirror) stopLoop() { |
| 979 | m.stopping.Store(true) |
| 980 | m.stopOnce.Do(func() { close(m.stop) }) |
| 981 | <-m.done |
| 982 | } |
| 983 | |
| 984 | // detach removes the mirror from the app registry and clears the sink hook. |
| 985 | func (m *takeoverMirror) detach() { |
| 986 | m.detachOnce.Do(func() { |
| 987 | m.app.takeoverMu.Lock() |
| 988 | if m.app.takeoverMirrors[m.key] == m { |
| 989 | delete(m.app.takeoverMirrors, m.key) |
| 990 | } |
| 991 | m.app.takeoverMu.Unlock() |
| 992 | if sink := m.currentSink(); sink != nil { |
| 993 | sink.setTakeoverMirror(nil) |
| 994 | } |
| 995 | }) |
| 996 | } |
| 997 | |
| 998 | // shutdown stops the forwarding loop and deregisters the mirror; when |
| 999 | // notifyServe is set (the mirror ended without a demotion, e.g. its tab |
| 1000 | // closed) Serve is told the writer is gone once it can accept that. |
| 1001 | func (m *takeoverMirror) stopAndFinalize(notifyServe bool) { |
| 1002 | m.stopLoop() |
| 1003 | m.detach() |
| 1004 | if notifyServe { |
| 1005 | m.mirrorEnd() |
| 1006 | } |
| 1007 | } |
| 1008 | |
| 1009 | func (m *takeoverMirror) currentSink() *tabEventSink { |
| 1010 | m.mu.Lock() |
| 1011 | defer m.mu.Unlock() |
| 1012 | return m.sink |
| 1013 | } |
| 1014 | |
| 1015 | // startSpectate keeps the demoted tab rendering by streaming Serve's frames |
| 1016 | // (the remote side is the writer again) into the tab's event channel. The |
| 1017 | // frames are the same wire contract the local reducer already consumes. |
| 1018 | func (m *takeoverMirror) startSpectate(tab *WorkspaceTab, sink *tabEventSink) { |
| 1019 | client, record, _, _ := m.snapshotClient() |
| 1020 | if tab == nil || sink == nil || client == nil { |
| 1021 | return |
| 1022 | } |
| 1023 | go func() { |
| 1024 | ctx, cancel := context.WithCancel(m.app.ctx) |
| 1025 | defer cancel() |
| 1026 | req, err := http.NewRequestWithContext(ctx, http.MethodGet, serveURL(record.base, "/events?all=1"), nil) |
| 1027 | if err != nil { |
| 1028 | return |
| 1029 | } |
| 1030 | req.Header.Set("Accept", "text/event-stream") |
| 1031 | resp, err := client.Do(req) |
| 1032 | if err != nil { |
| 1033 | return |
| 1034 | } |
| 1035 | defer resp.Body.Close() |
| 1036 | if resp.StatusCode != http.StatusOK { |
| 1037 | return |
| 1038 | } |
| 1039 | canonical := agent.CanonicalSessionPath(m.sessionPath) |
| 1040 | scanner := bufio.NewScanner(resp.Body) |
| 1041 | scanner.Buffer(make([]byte, 0, 64*1024), 4<<20) |
| 1042 | for scanner.Scan() { |
| 1043 | select { |
| 1044 | case <-m.app.ctx.Done(): |
| 1045 | return |
| 1046 | default: |
| 1047 | } |
| 1048 | line := scanner.Text() |
| 1049 | if !strings.HasPrefix(line, "data: ") { |
| 1050 | continue |
| 1051 | } |
| 1052 | var frame eventwire.Event |
| 1053 | if json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &frame) != nil { |
| 1054 | continue |
| 1055 | } |
| 1056 | if frame.SessionPath != canonical { |
| 1057 | continue |
| 1058 | } |
| 1059 | sink.emitRuntimeEvent(eventChannel, wireEventTab{Event: frame, TabID: m.tabIDSnapshot()}) |
| 1060 | } |
| 1061 | }() |
| 1062 | } |
| 1063 | |
| 1064 | func (m *takeoverMirror) tabIDSnapshot() string { |
| 1065 | m.mu.Lock() |
| 1066 | defer m.mu.Unlock() |
| 1067 | return m.tabID |
| 1068 | } |
| 1069 |