| 1 | package cli |
| 2 | |
| 3 | // Mirror side of CLI takeover: once a resident serve releases the lease the |
| 4 | // manager below owns the session, streams frames back to the remote tab that |
| 5 | // keeps watching read-only, and returns the lease when the tab reclaims it. |
| 6 | |
| 7 | import ( |
| 8 | "bytes" |
| 9 | "context" |
| 10 | "encoding/json" |
| 11 | "fmt" |
| 12 | "io" |
| 13 | "net/http" |
| 14 | "sync" |
| 15 | "sync/atomic" |
| 16 | "time" |
| 17 | |
| 18 | "reasonix/internal/agent" |
| 19 | "reasonix/internal/control" |
| 20 | "reasonix/internal/event" |
| 21 | "reasonix/internal/eventwire" |
| 22 | ) |
| 23 | |
| 24 | type cliTakeoverBinding struct { |
| 25 | path string |
| 26 | // canonical marks a final-format identity route ("session-id:<id>"): there |
| 27 | // is no path lease to move, ownership is the session directory's writer |
| 28 | // lock, released when the TUI yields and on process exit as a fallback. |
| 29 | canonical bool |
| 30 | record cliServeRecord |
| 31 | client *http.Client |
| 32 | grant cliTakeoverGrant |
| 33 | previous *control.SessionLeaseKeeper |
| 34 | priorMirror *cliTakeoverBinding |
| 35 | } |
| 36 | |
| 37 | const ( |
| 38 | cliTakeoverFlushEvery = 120 * time.Millisecond |
| 39 | cliTakeoverHeartbeat = 5 * time.Second |
| 40 | cliTakeoverMaxFrames = eventwire.MirrorBatchMaxFrames |
| 41 | ) |
| 42 | |
| 43 | const cliTakeoverRediscoverFailures = 3 |
| 44 | |
| 45 | type cliPendingReturn struct { |
| 46 | keeper *control.SessionLeaseKeeper |
| 47 | binding *cliTakeoverBinding |
| 48 | nextTry time.Time |
| 49 | backoff time.Duration |
| 50 | } |
| 51 | |
| 52 | // cliTakeoverManager is the outermost CLI event sink while a handed-off |
| 53 | // session is active. It preserves the terminal sink, mirrors the same typed |
| 54 | // frames to Serve, and cooperatively returns the lease when reclaim is seen. |
| 55 | // One manager survives controller rebuilds; AttachController updates its live |
| 56 | // authority pointer without replacing the sink wired into boot. |
| 57 | type cliTakeoverManager struct { |
| 58 | event.AuditForwarder |
| 59 | inner event.Sink |
| 60 | leases *control.SessionLeaseKeeper |
| 61 | |
| 62 | // Lock order is returnMu -> sendMu -> mu. Emit only takes mu, so the model |
| 63 | // event sink never waits for an HTTP request. |
| 64 | returnMu sync.Mutex |
| 65 | sendMu sync.Mutex |
| 66 | mu sync.Mutex |
| 67 | binding *cliTakeoverBinding |
| 68 | // yielded is the binding the last reclaim (or Close) returned. "/takeover |
| 69 | // takes it back" routes by the mirror's own key because a legacy path lease |
| 70 | // stays a path, so the controller's SessionRef cannot tell the kinds apart. |
| 71 | yielded *cliTakeoverBinding |
| 72 | revision uint64 |
| 73 | failures int |
| 74 | ctrl control.SessionAPI |
| 75 | queue eventwire.MirrorQueue |
| 76 | pending []*cliPendingReturn |
| 77 | // retirePending is a deterministic failure-injection seam for the pending |
| 78 | // return retry loop. Production calls RetireDetachedForHandoff directly. |
| 79 | retirePending func(*control.SessionLeaseKeeper, string, string) error |
| 80 | wake chan struct{} |
| 81 | stop chan struct{} |
| 82 | done chan struct{} |
| 83 | onYield func() |
| 84 | |
| 85 | started bool |
| 86 | stopOnce sync.Once |
| 87 | reclaiming atomic.Bool |
| 88 | returned atomic.Bool |
| 89 | closed atomic.Bool |
| 90 | } |
| 91 | |
| 92 | func newCLITakeoverManager(inner event.Sink, leases *control.SessionLeaseKeeper) *cliTakeoverManager { |
| 93 | return &cliTakeoverManager{AuditForwarder: event.AuditForwarder{Inner: inner}, inner: inner, leases: leases} |
| 94 | } |
| 95 | |
| 96 | func (m *cliTakeoverManager) SetInner(inner event.Sink) { |
| 97 | if m == nil { |
| 98 | return |
| 99 | } |
| 100 | m.mu.Lock() |
| 101 | m.inner = inner |
| 102 | m.Inner = inner |
| 103 | m.mu.Unlock() |
| 104 | } |
| 105 | |
| 106 | func (m *cliTakeoverManager) Emit(e event.Event) { |
| 107 | if m == nil { |
| 108 | return |
| 109 | } |
| 110 | m.mu.Lock() |
| 111 | inner := m.inner |
| 112 | m.mu.Unlock() |
| 113 | if inner != nil { |
| 114 | inner.Emit(e) |
| 115 | } |
| 116 | m.mu.Lock() |
| 117 | if m.binding != nil && !m.returned.Load() { |
| 118 | m.queue.Push(eventwire.ToWire(e)) |
| 119 | } |
| 120 | wake := m.wake |
| 121 | m.mu.Unlock() |
| 122 | if wake != nil { |
| 123 | select { |
| 124 | case wake <- struct{}{}: |
| 125 | default: |
| 126 | } |
| 127 | } |
| 128 | } |
| 129 | |
| 130 | func (m *cliTakeoverManager) EmitChecked(e event.Event) error { |
| 131 | m.mu.Lock() |
| 132 | inner := m.inner |
| 133 | m.mu.Unlock() |
| 134 | if checked, ok := inner.(event.CheckedSink); ok { |
| 135 | if err := checked.EmitChecked(e); err != nil { |
| 136 | return err |
| 137 | } |
| 138 | } else if inner != nil { |
| 139 | inner.Emit(e) |
| 140 | } |
| 141 | m.mu.Lock() |
| 142 | if m.binding != nil && !m.returned.Load() { |
| 143 | m.queue.Push(eventwire.ToWire(e)) |
| 144 | } |
| 145 | wake := m.wake |
| 146 | m.mu.Unlock() |
| 147 | if wake != nil { |
| 148 | select { |
| 149 | case wake <- struct{}{}: |
| 150 | default: |
| 151 | } |
| 152 | } |
| 153 | return nil |
| 154 | } |
| 155 | |
| 156 | func (m *cliTakeoverManager) AttachController(ctrl control.SessionAPI) { |
| 157 | if m == nil { |
| 158 | return |
| 159 | } |
| 160 | m.mu.Lock() |
| 161 | m.ctrl = ctrl |
| 162 | m.mu.Unlock() |
| 163 | } |
| 164 | |
| 165 | func (m *cliTakeoverManager) Activate(binding *cliTakeoverBinding) { |
| 166 | if m == nil || binding == nil { |
| 167 | return |
| 168 | } |
| 169 | m.returnMu.Lock() |
| 170 | defer m.returnMu.Unlock() |
| 171 | m.sendMu.Lock() |
| 172 | defer m.sendMu.Unlock() |
| 173 | m.mu.Lock() |
| 174 | m.binding = binding |
| 175 | m.yielded = nil |
| 176 | m.revision++ |
| 177 | m.failures = 0 |
| 178 | m.returned.Store(false) |
| 179 | m.reclaiming.Store(false) |
| 180 | m.ensureStartedLocked() |
| 181 | m.mu.Unlock() |
| 182 | } |
| 183 | |
| 184 | func (m *cliTakeoverManager) ensureStartedLocked() { |
| 185 | if m.started { |
| 186 | return |
| 187 | } |
| 188 | m.started = true |
| 189 | m.wake = make(chan struct{}, 1) |
| 190 | m.stop = make(chan struct{}) |
| 191 | m.done = make(chan struct{}) |
| 192 | go m.run() |
| 193 | } |
| 194 | |
| 195 | func (m *cliTakeoverManager) SetYieldCallback(fn func()) { |
| 196 | if m == nil { |
| 197 | return |
| 198 | } |
| 199 | m.mu.Lock() |
| 200 | m.onYield = fn |
| 201 | m.mu.Unlock() |
| 202 | } |
| 203 | |
| 204 | func (m *cliTakeoverManager) Reclaiming() bool { return m != nil && m.reclaiming.Load() } |
| 205 | func (m *cliTakeoverManager) Returned() bool { return m != nil && m.returned.Load() } |
| 206 | |
| 207 | // ResumeAfterYield clears the terminal-side marker after the TUI has acquired |
| 208 | // a different session. Returning a mirror ends that mirror permanently; the |
| 209 | // manager must not keep blocking the new session just because the process that |
| 210 | // hosted the old one stayed alive. |
| 211 | func (m *cliTakeoverManager) ResumeAfterYield() { |
| 212 | if m == nil { |
| 213 | return |
| 214 | } |
| 215 | m.returnMu.Lock() |
| 216 | defer m.returnMu.Unlock() |
| 217 | m.mu.Lock() |
| 218 | if m.binding == nil { |
| 219 | m.yielded = nil |
| 220 | m.returned.Store(false) |
| 221 | m.reclaiming.Store(false) |
| 222 | } |
| 223 | m.mu.Unlock() |
| 224 | } |
| 225 | |
| 226 | // yieldedBinding reports the mirror a reclaim returned, until the TUI either |
| 227 | // re-takes it or moves on to another session. |
| 228 | func (m *cliTakeoverManager) yieldedBinding() *cliTakeoverBinding { |
| 229 | if m == nil { |
| 230 | return nil |
| 231 | } |
| 232 | m.mu.Lock() |
| 233 | defer m.mu.Unlock() |
| 234 | return m.yielded |
| 235 | } |
| 236 | |
| 237 | func (m *cliTakeoverManager) snapshot() (*cliTakeoverBinding, control.SessionAPI, func(), uint64) { |
| 238 | m.mu.Lock() |
| 239 | defer m.mu.Unlock() |
| 240 | return m.binding, m.ctrl, m.onYield, m.revision |
| 241 | } |
| 242 | |
| 243 | func (m *cliTakeoverManager) run() { |
| 244 | m.mu.Lock() |
| 245 | wake, stop, done := m.wake, m.stop, m.done |
| 246 | m.mu.Unlock() |
| 247 | defer close(done) |
| 248 | timer := time.NewTimer(time.Hour) |
| 249 | if !timer.Stop() { |
| 250 | <-timer.C |
| 251 | } |
| 252 | defer timer.Stop() |
| 253 | heartbeat := time.NewTicker(cliTakeoverHeartbeat) |
| 254 | defer heartbeat.Stop() |
| 255 | retry := time.NewTicker(250 * time.Millisecond) |
| 256 | defer retry.Stop() |
| 257 | armed := false |
| 258 | for { |
| 259 | select { |
| 260 | case <-stop: |
| 261 | m.push(false) |
| 262 | return |
| 263 | case <-wake: |
| 264 | if !armed { |
| 265 | timer.Reset(cliTakeoverFlushEvery) |
| 266 | armed = true |
| 267 | } |
| 268 | case <-timer.C: |
| 269 | armed = false |
| 270 | if !m.push(false) { |
| 271 | return |
| 272 | } |
| 273 | case <-heartbeat.C: |
| 274 | if !m.push(true) { |
| 275 | return |
| 276 | } |
| 277 | case <-retry.C: |
| 278 | m.retryPendingReturns(false) |
| 279 | } |
| 280 | } |
| 281 | } |
| 282 | |
| 283 | func (m *cliTakeoverManager) drain() []eventwire.Event { |
| 284 | m.mu.Lock() |
| 285 | frames := m.queue.Take(cliTakeoverMaxFrames) |
| 286 | m.mu.Unlock() |
| 287 | return frames |
| 288 | } |
| 289 | |
| 290 | func (m *cliTakeoverManager) requeue(frames []eventwire.Event) { |
| 291 | if len(frames) == 0 { |
| 292 | return |
| 293 | } |
| 294 | m.mu.Lock() |
| 295 | m.queue.Prepend(frames) |
| 296 | m.mu.Unlock() |
| 297 | } |
| 298 | |
| 299 | func (m *cliTakeoverManager) wakeIfQueued() { |
| 300 | m.mu.Lock() |
| 301 | pending, wake := m.queue.Len() > 0, m.wake |
| 302 | m.mu.Unlock() |
| 303 | if pending && wake != nil { |
| 304 | select { |
| 305 | case wake <- struct{}{}: |
| 306 | default: |
| 307 | } |
| 308 | } |
| 309 | } |
| 310 | |
| 311 | func (m *cliTakeoverManager) push(heartbeat bool) bool { |
| 312 | m.sendMu.Lock() |
| 313 | defer m.sendMu.Unlock() |
| 314 | return m.pushLocked(heartbeat) |
| 315 | } |
| 316 | |
| 317 | func (m *cliTakeoverManager) pushLocked(heartbeat bool) bool { |
| 318 | if m.returned.Load() { |
| 319 | return false |
| 320 | } |
| 321 | binding, _, _, revision := m.snapshot() |
| 322 | if binding == nil || binding.client == nil || binding.grant.MirrorID == "" { |
| 323 | return true |
| 324 | } |
| 325 | frames := m.drain() |
| 326 | if len(frames) == 0 && !heartbeat { |
| 327 | return true |
| 328 | } |
| 329 | marshal := func(batch []eventwire.Event) ([]byte, error) { |
| 330 | return json.Marshal(map[string]any{ |
| 331 | "sessionPath": binding.path, "mirrorId": binding.grant.MirrorID, "frames": batch, |
| 332 | }) |
| 333 | } |
| 334 | batch, remainder, payload, marshalErr := eventwire.MarshalMirrorBatch(frames, eventwire.MirrorBatchMaxBytes, marshal) |
| 335 | if marshalErr == nil && len(batch) == 0 && len(frames) > 0 && len(remainder) > 0 { |
| 336 | remainder = remainder[1:] |
| 337 | } |
| 338 | m.requeue(remainder) |
| 339 | if marshalErr != nil { |
| 340 | m.requeue(batch) |
| 341 | return true |
| 342 | } |
| 343 | if len(batch) == 0 && len(frames) > 0 { |
| 344 | // A single frame larger than the HTTP protocol permits cannot ever be |
| 345 | // delivered. Durable history remains authoritative for its content. |
| 346 | m.wakeIfQueued() |
| 347 | if !heartbeat { |
| 348 | return true |
| 349 | } |
| 350 | payload, _ = marshal(nil) |
| 351 | } |
| 352 | ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) |
| 353 | req, err := http.NewRequestWithContext(ctx, http.MethodPost, binding.record.base+"/external/frames", bytes.NewReader(payload)) |
| 354 | if err == nil { |
| 355 | req.Header.Set("Content-Type", "application/json") |
| 356 | } |
| 357 | var resp *http.Response |
| 358 | if err == nil { |
| 359 | resp, err = binding.client.Do(req) |
| 360 | } |
| 361 | if err != nil { |
| 362 | cancel() |
| 363 | if !m.bindingCurrent(binding, revision) { |
| 364 | return true |
| 365 | } |
| 366 | m.requeue(batch) |
| 367 | return m.readoptLocked(binding, revision) |
| 368 | } |
| 369 | body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<16)) |
| 370 | resp.Body.Close() |
| 371 | cancel() |
| 372 | if !m.bindingCurrent(binding, revision) { |
| 373 | return true |
| 374 | } |
| 375 | if resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden || resp.StatusCode == http.StatusConflict { |
| 376 | m.requeue(batch) |
| 377 | return m.readoptLocked(binding, revision) |
| 378 | } |
| 379 | if resp.StatusCode != http.StatusOK { |
| 380 | m.requeue(batch) |
| 381 | m.mu.Lock() |
| 382 | if m.binding == binding && m.revision == revision { |
| 383 | m.failures++ |
| 384 | } |
| 385 | failures := m.failures |
| 386 | m.mu.Unlock() |
| 387 | if failures >= cliTakeoverRediscoverFailures { |
| 388 | return m.readoptLocked(binding, revision) |
| 389 | } |
| 390 | return true |
| 391 | } |
| 392 | m.mu.Lock() |
| 393 | if m.binding == binding && m.revision == revision { |
| 394 | m.failures = 0 |
| 395 | } |
| 396 | m.mu.Unlock() |
| 397 | var out struct { |
| 398 | ReclaimRequested bool `json:"reclaimRequested"` |
| 399 | ReclaimMode string `json:"reclaimMode"` |
| 400 | } |
| 401 | if json.Unmarshal(body, &out) == nil && out.ReclaimRequested { |
| 402 | m.requestYieldFor(binding, revision, out.ReclaimMode == "interrupt") |
| 403 | return true |
| 404 | } |
| 405 | m.wakeIfQueued() |
| 406 | return true |
| 407 | } |
| 408 | |
| 409 | func (m *cliTakeoverManager) bindingCurrent(binding *cliTakeoverBinding, revision uint64) bool { |
| 410 | m.mu.Lock() |
| 411 | defer m.mu.Unlock() |
| 412 | return m.binding == binding && m.revision == revision && !m.returned.Load() |
| 413 | } |
| 414 | |
| 415 | // readoptLocked discovers the current Serve endpoint and rotates the mirror |
| 416 | // generation. sendMu is held by the caller, so a binding switch cannot race a |
| 417 | // late response from the endpoint being replaced. |
| 418 | func (m *cliTakeoverManager) readoptLocked(binding *cliTakeoverBinding, revision uint64) bool { |
| 419 | if binding == nil || !m.bindingCurrent(binding, revision) { |
| 420 | return false |
| 421 | } |
| 422 | conflicted := false |
| 423 | for _, record := range discoverCLIServesForTakeover() { |
| 424 | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 425 | client, err := cliServeClient(ctx, record) |
| 426 | if err != nil { |
| 427 | cancel() |
| 428 | continue |
| 429 | } |
| 430 | payload, _ := json.Marshal(map[string]string{"sessionPath": binding.path, "writerId": agent.SessionWriterID()}) |
| 431 | req, err := http.NewRequestWithContext(ctx, http.MethodPost, record.base+"/adopt", bytes.NewReader(payload)) |
| 432 | if err == nil { |
| 433 | req.Header.Set("Content-Type", "application/json") |
| 434 | } |
| 435 | var resp *http.Response |
| 436 | if err == nil { |
| 437 | resp, err = client.Do(req) |
| 438 | } |
| 439 | if err != nil { |
| 440 | cancel() |
| 441 | continue |
| 442 | } |
| 443 | body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<16)) |
| 444 | resp.Body.Close() |
| 445 | cancel() |
| 446 | if resp.StatusCode == http.StatusConflict { |
| 447 | conflicted = true |
| 448 | continue |
| 449 | } |
| 450 | var grant cliTakeoverGrant |
| 451 | if resp.StatusCode == http.StatusOK && json.Unmarshal(body, &grant) == nil && |
| 452 | grant.MirrorID != "" && grant.ReturnHandoffID != "" && grant.SourceWriterID != "" && |
| 453 | grant.TargetWriterID == agent.SessionWriterID() && |
| 454 | agent.CanonicalSessionPath(grant.SessionPath) == agent.CanonicalSessionPath(binding.path) { |
| 455 | m.mu.Lock() |
| 456 | if m.binding == binding && m.revision == revision && !m.returned.Load() { |
| 457 | next := *binding |
| 458 | next.record, next.client, next.grant = record, client, grant |
| 459 | m.binding = &next |
| 460 | m.revision++ |
| 461 | m.failures = 0 |
| 462 | } |
| 463 | m.mu.Unlock() |
| 464 | return true |
| 465 | } |
| 466 | } |
| 467 | if conflicted { |
| 468 | m.requestYieldFor(binding, revision, false) |
| 469 | } |
| 470 | return true |
| 471 | } |
| 472 | |
| 473 | func (m *cliTakeoverManager) requestYieldFor(binding *cliTakeoverBinding, revision uint64, interrupt bool) { |
| 474 | if !m.bindingCurrent(binding, revision) { |
| 475 | return |
| 476 | } |
| 477 | if !m.reclaiming.CompareAndSwap(false, true) { |
| 478 | return |
| 479 | } |
| 480 | current, ctrl, callback, currentRevision := m.snapshot() |
| 481 | if current != binding || currentRevision != revision { |
| 482 | m.reclaiming.Store(false) |
| 483 | return |
| 484 | } |
| 485 | if interrupt && ctrl != nil { |
| 486 | ctrl.Cancel() |
| 487 | } |
| 488 | go func() { |
| 489 | deadline := time.Now().Add(cliTakeoverTimeout) |
| 490 | for cliControllerHasActiveRuntimeWork(ctrl) && time.Now().Before(deadline) { |
| 491 | time.Sleep(50 * time.Millisecond) |
| 492 | } |
| 493 | if cliControllerHasActiveRuntimeWork(ctrl) { |
| 494 | m.reclaiming.Store(false) |
| 495 | return |
| 496 | } |
| 497 | if err := m.returnLeaseFor(binding, revision); err != nil { |
| 498 | m.reclaiming.Store(false) |
| 499 | return |
| 500 | } |
| 501 | if callback != nil { |
| 502 | callback() |
| 503 | } |
| 504 | }() |
| 505 | } |
| 506 | |
| 507 | func (m *cliTakeoverManager) returnLease() error { |
| 508 | return m.returnLeaseFor(nil, 0) |
| 509 | } |
| 510 | |
| 511 | func (m *cliTakeoverManager) returnLeaseFor(expected *cliTakeoverBinding, revision uint64) error { |
| 512 | expectedPath := "" |
| 513 | if expected != nil { |
| 514 | if !m.bindingCurrent(expected, revision) { |
| 515 | current, _, _, _ := m.snapshot() |
| 516 | if current == nil || agent.CanonicalSessionPath(current.path) != agent.CanonicalSessionPath(expected.path) { |
| 517 | return fmt.Errorf("takeover mirror changed before reclaim completed") |
| 518 | } |
| 519 | } |
| 520 | expectedPath = expected.path |
| 521 | } |
| 522 | return m.returnMirrorTransaction(expectedPath, true, true, func(current *cliTakeoverBinding) error { |
| 523 | if current.canonical { |
| 524 | // The canonical writer lock is released by the live TUI handoff; the |
| 525 | // flushed snapshot above is the only durable step the reservation covered. |
| 526 | return nil |
| 527 | } |
| 528 | return m.leases.ReleaseForHandoff(current.grant.SourceWriterID, current.grant.ReturnHandoffID) |
| 529 | }) |
| 530 | } |
| 531 | |
| 532 | // RebindAway acquires a new ordinary session before returning the mirrored |
| 533 | // one. It lets /resume and related TUI switches keep their original failure |
| 534 | // atomicity while still honoring Serve's reverse reservation. |
| 535 | func (m *cliTakeoverManager) RebindAway(path string) (bool, error) { |
| 536 | if m == nil { |
| 537 | return false, nil |
| 538 | } |
| 539 | binding, _, _, _ := m.snapshot() |
| 540 | if binding == nil || m.returned.Load() || agent.CanonicalSessionPath(binding.path) == agent.CanonicalSessionPath(path) { |
| 541 | return false, nil |
| 542 | } |
| 543 | err := m.returnCurrentMirror(binding.path, func(current *cliTakeoverBinding) error { |
| 544 | if current.canonical { |
| 545 | return nil |
| 546 | } |
| 547 | return m.leases.RebindReturningCurrent(path, current.grant.SourceWriterID, current.grant.ReturnHandoffID) |
| 548 | }) |
| 549 | return true, err |
| 550 | } |
| 551 | |
| 552 | // cliAcquireFreeSession starts an ordinary failure-atomic switch. The newly |
| 553 | // acquired target stays in leases while the source binding remains detached |
| 554 | // and live until the caller has loaded and authorized the candidate session. |
| 555 | func cliAcquireFreeSession(path string, leases *control.SessionLeaseKeeper, manager *cliTakeoverManager) (*cliTakeoverBinding, error) { |
| 556 | if leases == nil { |
| 557 | return &cliTakeoverBinding{path: path}, nil |
| 558 | } |
| 559 | if manager != nil && manager.Reclaiming() { |
| 560 | return nil, fmt.Errorf("the remote side is reclaiming the current session") |
| 561 | } |
| 562 | binding := &cliTakeoverBinding{path: path} |
| 563 | if manager != nil { |
| 564 | current, _, _, _ := manager.snapshot() |
| 565 | if current != nil && !manager.Returned() && agent.CanonicalSessionPath(current.path) != agent.CanonicalSessionPath(path) { |
| 566 | binding.priorMirror = current |
| 567 | } |
| 568 | } |
| 569 | previous, err := leases.RebindDetaching(path) |
| 570 | if err != nil { |
| 571 | return nil, err |
| 572 | } |
| 573 | binding.previous = previous |
| 574 | return binding, nil |
| 575 | } |
| 576 | |
| 577 | // cliPrepareTakeoverCandidate reloads after acquisition. For a Serve handoff, |
| 578 | // this observes the Snapshot completed by /handoff rather than the stale |
| 579 | // preflight view. Authority is bound to the private candidate before the |
| 580 | // controller publishes it through Resume. |
| 581 | func cliPrepareTakeoverCandidate(binding *cliTakeoverBinding, leases *control.SessionLeaseKeeper) (*agent.Session, error) { |
| 582 | if binding == nil { |
| 583 | return nil, fmt.Errorf("takeover binding unavailable") |
| 584 | } |
| 585 | loaded, err := loadResumableSession(binding.path) |
| 586 | if err != nil { |
| 587 | return nil, err |
| 588 | } |
| 589 | if leases != nil { |
| 590 | if err := leases.BindSessionAuthority(loaded); err != nil { |
| 591 | return nil, err |
| 592 | } |
| 593 | } |
| 594 | return loaded, nil |
| 595 | } |
| 596 | |
| 597 | // commitPrevious retires the source keeper only after the handed-off target |
| 598 | // has been loaded successfully. A mirrored source is returned through the |
| 599 | // manager's single leave step; an ordinary source is simply released. |
| 600 | func (b *cliTakeoverBinding) commitPrevious(manager *cliTakeoverManager) error { |
| 601 | if b == nil { |
| 602 | return nil |
| 603 | } |
| 604 | if b.priorMirror == nil { |
| 605 | b.previous.RetireDetached() |
| 606 | b.previous = nil |
| 607 | return nil |
| 608 | } |
| 609 | if manager == nil { |
| 610 | return fmt.Errorf("takeover manager unavailable for mirrored source") |
| 611 | } |
| 612 | return manager.commitPriorMirror(b) |
| 613 | } |
| 614 | |
| 615 | func (m *cliTakeoverManager) commitPriorMirror(next *cliTakeoverBinding) error { |
| 616 | if m == nil || next == nil || next.priorMirror == nil { |
| 617 | return nil |
| 618 | } |
| 619 | if err := m.leaveMirror(next.priorMirror.path, next.previous); err != nil { |
| 620 | return err |
| 621 | } |
| 622 | next.previous = nil |
| 623 | return nil |
| 624 | } |
| 625 | |
| 626 | // leaveMirror returns the active mirror on behalf of a session switch that |
| 627 | // has already secured its target, so no frame of the next session travels |
| 628 | // under the old mirror id and the remote tab regains its writer. It is the one |
| 629 | // exit every switch takes before binding the next session — /resume and |
| 630 | // /takeover, legacy and canonical targets alike — and it retires the source's |
| 631 | // ownership by kind. A legacy lease publishes its reverse reservation from |
| 632 | // whichever keeper still holds it: previous when a legacy switch already moved |
| 633 | // the source out of the live keeper, otherwise the live keeper; a lease nobody |
| 634 | // holds (an exclusive-mode import released its compatibility lease) has |
| 635 | // nothing to publish. A canonical identity's writer lock travels with the |
| 636 | // controller binding the caller moves, so only the mirror itself ends. |
| 637 | // expectedPath, when set, names the mirror the caller observed; a different |
| 638 | // current mirror aborts the switch. |
| 639 | func (m *cliTakeoverManager) leaveMirror(expectedPath string, previous *control.SessionLeaseKeeper) error { |
| 640 | if m == nil { |
| 641 | return nil |
| 642 | } |
| 643 | current, _, _, _ := m.snapshot() |
| 644 | if current == nil || m.returned.Load() { |
| 645 | // Nothing is mirrored, so the detached source keeper is plain state. |
| 646 | previous.RetireDetached() |
| 647 | return nil |
| 648 | } |
| 649 | if expectedPath == "" { |
| 650 | expectedPath = current.path |
| 651 | } |
| 652 | return m.returnCurrentMirror(expectedPath, func(current *cliTakeoverBinding) error { |
| 653 | if current.canonical { |
| 654 | previous.RetireDetached() |
| 655 | return nil |
| 656 | } |
| 657 | holder := previous |
| 658 | if holder == nil { |
| 659 | holder = m.leases |
| 660 | } |
| 661 | if holder.HeldPath() != agent.CanonicalSessionPath(current.path) { |
| 662 | previous.RetireDetached() |
| 663 | return nil |
| 664 | } |
| 665 | if previous != nil { |
| 666 | return previous.RetireDetachedForHandoff(current.grant.SourceWriterID, current.grant.ReturnHandoffID) |
| 667 | } |
| 668 | return m.leases.ReleaseForHandoff(current.grant.SourceWriterID, current.grant.ReturnHandoffID) |
| 669 | }) |
| 670 | } |
| 671 | |
| 672 | func (m *cliTakeoverManager) mirrorEnd(binding *cliTakeoverBinding) { |
| 673 | if m == nil { |
| 674 | return |
| 675 | } |
| 676 | m.sendMu.Lock() |
| 677 | defer m.sendMu.Unlock() |
| 678 | m.mirrorEndLocked(binding) |
| 679 | } |
| 680 | |
| 681 | func (m *cliTakeoverManager) mirrorEndLocked(binding *cliTakeoverBinding) { |
| 682 | if binding == nil || binding.client == nil { |
| 683 | return |
| 684 | } |
| 685 | payload, _ := json.Marshal(map[string]string{"sessionPath": binding.path, "mirrorId": binding.grant.MirrorID}) |
| 686 | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 687 | defer cancel() |
| 688 | req, err := http.NewRequestWithContext(ctx, http.MethodPost, binding.record.base+"/mirror-end", bytes.NewReader(payload)) |
| 689 | if err != nil { |
| 690 | return |
| 691 | } |
| 692 | req.Header.Set("Content-Type", "application/json") |
| 693 | resp, err := binding.client.Do(req) |
| 694 | if err == nil { |
| 695 | _, _ = io.Copy(io.Discard, resp.Body) |
| 696 | resp.Body.Close() |
| 697 | } |
| 698 | } |
| 699 | |
| 700 | func (m *cliTakeoverManager) holdPendingReturn(keeper *control.SessionLeaseKeeper, binding *cliTakeoverBinding) { |
| 701 | if m == nil || keeper == nil || binding == nil { |
| 702 | return |
| 703 | } |
| 704 | m.mu.Lock() |
| 705 | m.pending = append(m.pending, &cliPendingReturn{ |
| 706 | keeper: keeper, binding: binding, nextTry: time.Now().Add(200 * time.Millisecond), backoff: 200 * time.Millisecond, |
| 707 | }) |
| 708 | m.ensureStartedLocked() |
| 709 | wake := m.wake |
| 710 | m.mu.Unlock() |
| 711 | if wake != nil { |
| 712 | select { |
| 713 | case wake <- struct{}{}: |
| 714 | default: |
| 715 | } |
| 716 | } |
| 717 | } |
| 718 | |
| 719 | func (m *cliTakeoverManager) retryPendingReturns(force bool) { |
| 720 | if m == nil { |
| 721 | return |
| 722 | } |
| 723 | if force { |
| 724 | m.returnMu.Lock() |
| 725 | } else if !m.returnMu.TryLock() { |
| 726 | // The active mirror return transaction owns the forwarding loop. Do not |
| 727 | // strand that transaction while it joins this loop; the next retry tick |
| 728 | // will pick these detached keepers up. |
| 729 | return |
| 730 | } |
| 731 | defer m.returnMu.Unlock() |
| 732 | now := time.Now() |
| 733 | m.mu.Lock() |
| 734 | pending := append([]*cliPendingReturn(nil), m.pending...) |
| 735 | m.mu.Unlock() |
| 736 | for _, item := range pending { |
| 737 | if item == nil || item.keeper == nil || item.binding == nil || (!force && now.Before(item.nextTry)) { |
| 738 | continue |
| 739 | } |
| 740 | var err error |
| 741 | if m.retirePending != nil { |
| 742 | err = m.retirePending(item.keeper, item.binding.grant.SourceWriterID, item.binding.grant.ReturnHandoffID) |
| 743 | } else { |
| 744 | err = item.keeper.RetireDetachedForHandoff(item.binding.grant.SourceWriterID, item.binding.grant.ReturnHandoffID) |
| 745 | } |
| 746 | if err == nil { |
| 747 | m.mirrorEnd(item.binding) |
| 748 | m.mu.Lock() |
| 749 | for i, candidate := range m.pending { |
| 750 | if candidate == item { |
| 751 | m.pending = append(m.pending[:i], m.pending[i+1:]...) |
| 752 | break |
| 753 | } |
| 754 | } |
| 755 | m.mu.Unlock() |
| 756 | continue |
| 757 | } |
| 758 | m.mu.Lock() |
| 759 | item.backoff = min(item.backoff*2, 5*time.Second) |
| 760 | item.nextTry = now.Add(item.backoff) |
| 761 | m.mu.Unlock() |
| 762 | } |
| 763 | } |
| 764 | |
| 765 | // cliReturnFailedTakeover restores the source binding after a candidate load |
| 766 | // or commit failure. A failed reverse-reservation write leaves the target in a |
| 767 | // manager-owned detached keeper; mirror-end is withheld until a retry succeeds. |
| 768 | func cliReturnFailedTakeover(binding *cliTakeoverBinding, leases *control.SessionLeaseKeeper, manager *cliTakeoverManager) error { |
| 769 | if binding == nil || leases == nil { |
| 770 | return nil |
| 771 | } |
| 772 | if binding.grant.ReturnHandoffID != "" && binding.grant.SourceWriterID != "" { |
| 773 | var pending *control.SessionLeaseKeeper |
| 774 | var err error |
| 775 | if binding.previous != nil { |
| 776 | pending, err = leases.RestoreDetachedReturningCurrent( |
| 777 | binding.previous, binding.grant.SourceWriterID, binding.grant.ReturnHandoffID, |
| 778 | ) |
| 779 | binding.previous = nil |
| 780 | } else { |
| 781 | pending = leases.Split() |
| 782 | if pending == nil { |
| 783 | return fmt.Errorf("failed takeover target lease is unavailable") |
| 784 | } |
| 785 | err = pending.RetireDetachedForHandoff(binding.grant.SourceWriterID, binding.grant.ReturnHandoffID) |
| 786 | if err == nil { |
| 787 | pending = nil |
| 788 | } |
| 789 | } |
| 790 | if err != nil { |
| 791 | if manager == nil || pending == nil { |
| 792 | return fmt.Errorf("return failed takeover lease: %w", err) |
| 793 | } |
| 794 | manager.holdPendingReturn(pending, binding) |
| 795 | return fmt.Errorf("return failed takeover lease (retrying): %w", err) |
| 796 | } |
| 797 | } else { |
| 798 | current := leases.Split() |
| 799 | if binding.previous != nil { |
| 800 | leases.Adopt(binding.previous) |
| 801 | binding.previous = nil |
| 802 | } |
| 803 | if current != nil { |
| 804 | current.RetireDetached() |
| 805 | } |
| 806 | } |
| 807 | if binding.grant.MirrorID == "" { |
| 808 | return nil |
| 809 | } |
| 810 | if manager != nil { |
| 811 | manager.mirrorEnd(binding) |
| 812 | } else { |
| 813 | (&cliTakeoverManager{}).mirrorEnd(binding) |
| 814 | } |
| 815 | return nil |
| 816 | } |
| 817 | |
| 818 | func cliEndFailedHandoff(binding *cliTakeoverBinding) { |
| 819 | if binding == nil { |
| 820 | return |
| 821 | } |
| 822 | m := &cliTakeoverManager{} |
| 823 | m.mirrorEnd(binding) |
| 824 | } |
| 825 | |
| 826 | // Close returns an active mirrored session on ordinary CLI exit. A concurrent |
| 827 | // reclaim owns the same transaction; wait for it rather than publishing a |
| 828 | // second reservation. |
| 829 | func (m *cliTakeoverManager) Close() error { |
| 830 | if m == nil { |
| 831 | return nil |
| 832 | } |
| 833 | if !m.closed.CompareAndSwap(false, true) { |
| 834 | return nil |
| 835 | } |
| 836 | var closeErr error |
| 837 | if m.reclaiming.Load() { |
| 838 | _, ctrl, _, _ := m.snapshot() |
| 839 | deadline := time.Now().Add(cliTakeoverTimeout) |
| 840 | for cliControllerHasActiveRuntimeWork(ctrl) && time.Now().Before(deadline) { |
| 841 | time.Sleep(50 * time.Millisecond) |
| 842 | } |
| 843 | if cliControllerHasActiveRuntimeWork(ctrl) { |
| 844 | return fmt.Errorf("timed out waiting for the active turn to yield its session") |
| 845 | } |
| 846 | } |
| 847 | if err := m.returnLease(); err != nil { |
| 848 | closeErr = err |
| 849 | } |
| 850 | m.mu.Lock() |
| 851 | started, stop, done := m.started, m.stop, m.done |
| 852 | m.mu.Unlock() |
| 853 | if started { |
| 854 | m.stopOnce.Do(func() { close(stop) }) |
| 855 | <-done |
| 856 | } |
| 857 | m.retryPendingReturns(true) |
| 858 | m.mu.Lock() |
| 859 | remaining := append([]*cliPendingReturn(nil), m.pending...) |
| 860 | m.pending = nil |
| 861 | m.mu.Unlock() |
| 862 | for _, item := range remaining { |
| 863 | if item != nil && item.keeper != nil { |
| 864 | // Close is the final CLI teardown. Keep the target fenced until this |
| 865 | // point, then let process cleanup release an unreturnable OS lease. |
| 866 | item.keeper.Release() |
| 867 | } |
| 868 | } |
| 869 | if len(remaining) > 0 && closeErr == nil { |
| 870 | closeErr = fmt.Errorf("unable to publish %d pending session return reservation(s)", len(remaining)) |
| 871 | } |
| 872 | return closeErr |
| 873 | } |
| 874 |