feat(daemon): park authenticated remote routes
- commit
- 9bd61851a0d009d293b771fe39729dba1552f99f
- author
- brice · · 2026-08-25 13:35 UTC
- parent
- e40f2b0
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
9 files · +903 -7
M internal/usecase/daemon/client.go +12 -1
| @@ -100,6 +100,10 @@ type attachedClient struct { | |||
| 100 | 100 | pendingRouteIdentity bool | |
| 101 | 101 | samePeerOfferMu sync.Mutex | |
| 102 | 102 | samePeerOffer *ports.ExactSessionTarget | |
| 103 | + | parkedRouteMu sync.Mutex | |
| 104 | + | parkedRoute *parkedRouteLease | |
| 105 | + | parkedRouteOutput atomic.Bool | |
| 106 | + | parkedRouteFullPending atomic.Bool | |
| 103 | 107 | routeAttentionSubscription ports.RouteAttentionSubscription | |
| 104 | 108 | // routeCreatedSession marks a session created by this attachment's route. | |
| 105 | 109 | // A handshake that never commits Welcome must tear down that exact empty | |
| @@ -514,6 +518,9 @@ func (d *Daemon) boundedSendOutputErrTransport(ac *attachedClient, b []byte) (po | |||
| 514 | 518 | if !ac.transportSnapshotCurrent(expected) { | |
| 515 | 519 | return expected.transport, errTransportReplaced | |
| 516 | 520 | } | |
| 521 | + | if ac.parkedRouteOutput.Load() || ac.parkedRouteFullPending.Load() { | |
| 522 | + | return expected.transport, nil | |
| 523 | + | } | |
| 517 | 524 | ac.output.lockView() | |
| 518 | 525 | defer ac.output.unlockView() | |
| 519 | 526 | frame, err := ac.output.sideEffectLocked(b, ac.echoAck.Load()) | |
| @@ -528,6 +535,9 @@ func (d *Daemon) boundedSendOutputErrTransport(ac *attachedClient, b []byte) (po | |||
| 528 | 535 | if !ac.transportSnapshotCurrent(expected) { | |
| 529 | 536 | return errTransportReplaced | |
| 530 | 537 | } | |
| 538 | + | if ac.parkedRouteOutput.Load() || ac.parkedRouteFullPending.Load() { | |
| 539 | + | return nil | |
| 540 | + | } | |
| 531 | 541 | ac.output.lockView() | |
| 532 | 542 | defer ac.output.unlockView() | |
| 533 | 543 | frame, err := ac.output.sideEffectLocked(b, ac.echoAck.Load()) | |
| @@ -784,7 +794,8 @@ func (d *Daemon) ensureAttachmentRenderCoordinatorPrelocked(entry *session) *ren | |||
| 784 | 794 | // outputStateStream publishes capacity atomically. Do not take | |
| 785 | 795 | // attached.sendMu here: a slow transport may be holding it for an | |
| 786 | 796 | // in-flight Send, and that peer must not gate healthy attachments. | |
| 787 | - | return attached == nil || attached.output == nil || !attached.output.atCapacity() | |
| 797 | + | return attached == nil || | |
| 798 | + | (!attached.parkedRouteOutput.Load() && (attached.output == nil || !attached.output.atCapacity())) | |
| 788 | 799 | }, | |
| 789 | 800 | }) | |
| 790 | 801 | installAttachmentRenderCoordinator(entry, rc) | |
| @@ -100,6 +100,10 @@ type attachedClient struct { | |||||
| 100 | pendingRouteIdentity bool | 100 | pendingRouteIdentity bool | ||
| 101 | samePeerOfferMu sync.Mutex | 101 | samePeerOfferMu sync.Mutex | ||
| 102 | samePeerOffer *ports.ExactSessionTarget | 102 | samePeerOffer *ports.ExactSessionTarget | ||
| 103 | + | parkedRouteMu sync.Mutex | |||
| 104 | + | parkedRoute *parkedRouteLease | |||
| 105 | + | parkedRouteOutput atomic.Bool | |||
| 106 | + | parkedRouteFullPending atomic.Bool | |||
| 103 | routeAttentionSubscription ports.RouteAttentionSubscription | 107 | routeAttentionSubscription ports.RouteAttentionSubscription | ||
| 104 | // routeCreatedSession marks a session created by this attachment's route. | 108 | // routeCreatedSession marks a session created by this attachment's route. | ||
| 105 | // A handshake that never commits Welcome must tear down that exact empty | 109 | // A handshake that never commits Welcome must tear down that exact empty | ||
| @@ -514,6 +518,9 @@ func (d *Daemon) boundedSendOutputErrTransport(ac *attachedClient, b []byte) (po | |||||
| 514 | if !ac.transportSnapshotCurrent(expected) { | 518 | if !ac.transportSnapshotCurrent(expected) { | ||
| 515 | return expected.transport, errTransportReplaced | 519 | return expected.transport, errTransportReplaced | ||
| 516 | } | 520 | } | ||
| 521 | + | if ac.parkedRouteOutput.Load() || ac.parkedRouteFullPending.Load() { | |||
| 522 | + | return expected.transport, nil | |||
| 523 | + | } | |||
| 517 | ac.output.lockView() | 524 | ac.output.lockView() | ||
| 518 | defer ac.output.unlockView() | 525 | defer ac.output.unlockView() | ||
| 519 | frame, err := ac.output.sideEffectLocked(b, ac.echoAck.Load()) | 526 | frame, err := ac.output.sideEffectLocked(b, ac.echoAck.Load()) | ||
| @@ -528,6 +535,9 @@ func (d *Daemon) boundedSendOutputErrTransport(ac *attachedClient, b []byte) (po | |||||
| 528 | if !ac.transportSnapshotCurrent(expected) { | 535 | if !ac.transportSnapshotCurrent(expected) { | ||
| 529 | return errTransportReplaced | 536 | return errTransportReplaced | ||
| 530 | } | 537 | } | ||
| 538 | + | if ac.parkedRouteOutput.Load() || ac.parkedRouteFullPending.Load() { | |||
| 539 | + | return nil | |||
| 540 | + | } | |||
| 531 | ac.output.lockView() | 541 | ac.output.lockView() | ||
| 532 | defer ac.output.unlockView() | 542 | defer ac.output.unlockView() | ||
| 533 | frame, err := ac.output.sideEffectLocked(b, ac.echoAck.Load()) | 543 | frame, err := ac.output.sideEffectLocked(b, ac.echoAck.Load()) | ||
| @@ -784,7 +794,8 @@ func (d *Daemon) ensureAttachmentRenderCoordinatorPrelocked(entry *session) *ren | |||||
| 784 | // outputStateStream publishes capacity atomically. Do not take | 794 | // outputStateStream publishes capacity atomically. Do not take | ||
| 785 | // attached.sendMu here: a slow transport may be holding it for an | 795 | // attached.sendMu here: a slow transport may be holding it for an | ||
| 786 | // in-flight Send, and that peer must not gate healthy attachments. | 796 | // in-flight Send, and that peer must not gate healthy attachments. | ||
| 787 | - | return attached == nil || attached.output == nil || !attached.output.atCapacity() | 797 | + | return attached == nil || |
| 798 | + | (!attached.parkedRouteOutput.Load() && (attached.output == nil || !attached.output.atCapacity())) | |||
| 788 | }, | 799 | }, | ||
| 789 | }) | 800 | }) | ||
| 790 | installAttachmentRenderCoordinator(entry, rc) | 801 | installAttachmentRenderCoordinator(entry, rc) | ||
M internal/usecase/daemon/client_frame_routing.go +7 -0
| @@ -133,6 +133,13 @@ func (d *Daemon) handleAttachmentClientFrame(token attachmentConnectionToken, f | |||
| 133 | 133 | break | |
| 134 | 134 | } | |
| 135 | 135 | d.switchSamePeerForAttachment(token, request) | |
| 136 | + | case ports.MsgParkedRouteRequest: | |
| 137 | + | request, derr := ports.UnmarshalParkedRouteRequest(f.Payload) | |
| 138 | + | if derr != nil { | |
| 139 | + | d.log.Warn("malformed parked-route request", "err", derr) | |
| 140 | + | break | |
| 141 | + | } | |
| 142 | + | d.handleParkedRouteRequest(token, request) | |
| 136 | 143 | case ports.MsgRecentRouteSnapshot: | |
| 137 | 144 | snapshot, derr := ports.UnmarshalRecentRouteSnapshot(f.Payload) | |
| 138 | 145 | if derr != nil { | |
| @@ -133,6 +133,13 @@ func (d *Daemon) handleAttachmentClientFrame(token attachmentConnectionToken, f | |||||
| 133 | break | 133 | break | ||
| 134 | } | 134 | } | ||
| 135 | d.switchSamePeerForAttachment(token, request) | 135 | d.switchSamePeerForAttachment(token, request) | ||
| 136 | + | case ports.MsgParkedRouteRequest: | |||
| 137 | + | request, derr := ports.UnmarshalParkedRouteRequest(f.Payload) | |||
| 138 | + | if derr != nil { | |||
| 139 | + | d.log.Warn("malformed parked-route request", "err", derr) | |||
| 140 | + | break | |||
| 141 | + | } | |||
| 142 | + | d.handleParkedRouteRequest(token, request) | |||
| 136 | case ports.MsgRecentRouteSnapshot: | 143 | case ports.MsgRecentRouteSnapshot: | ||
| 137 | snapshot, derr := ports.UnmarshalRecentRouteSnapshot(f.Payload) | 144 | snapshot, derr := ports.UnmarshalRecentRouteSnapshot(f.Payload) | ||
| 138 | if derr != nil { | 145 | if derr != nil { | ||
M internal/usecase/daemon/client_transport_errors.go +2 -0
| @@ -66,6 +66,7 @@ func (d *Daemon) finishClientGone(sess *session, ac *attachedClient, failed port | |||
| 66 | 66 | return | |
| 67 | 67 | } | |
| 68 | 68 | ac.clearSamePeerOffer() | |
| 69 | + | ac.clearParkedRoute() | |
| 69 | 70 | if d.afterClientGoneDetach != nil { | |
| 70 | 71 | d.afterClientGoneDetach() | |
| 71 | 72 | } | |
| @@ -171,6 +172,7 @@ func (d *Daemon) reserveAttachmentSendErrorCleanup(token attachmentConnectionTok | |||
| 171 | 172 | ||
| 172 | 173 | func (d *Daemon) finishSendErrorDetach(sess *session, ac *attachedClient, failed ports.Transport) { | |
| 173 | 174 | ac.clearSamePeerOffer() | |
| 175 | + | ac.clearParkedRoute() | |
| 174 | 176 | name := sess.nameSnapshot() | |
| 175 | 177 | if rc := sess.renderCoordinator(); rc != nil { | |
| 176 | 178 | rc.noteDetach(ac) | |
| @@ -66,6 +66,7 @@ func (d *Daemon) finishClientGone(sess *session, ac *attachedClient, failed port | |||||
| 66 | return | 66 | return | ||
| 67 | } | 67 | } | ||
| 68 | ac.clearSamePeerOffer() | 68 | ac.clearSamePeerOffer() | ||
| 69 | + | ac.clearParkedRoute() | |||
| 69 | if d.afterClientGoneDetach != nil { | 70 | if d.afterClientGoneDetach != nil { | ||
| 70 | d.afterClientGoneDetach() | 71 | d.afterClientGoneDetach() | ||
| 71 | } | 72 | } | ||
| @@ -171,6 +172,7 @@ func (d *Daemon) reserveAttachmentSendErrorCleanup(token attachmentConnectionTok | |||||
| 171 | 172 | ||||
| 172 | func (d *Daemon) finishSendErrorDetach(sess *session, ac *attachedClient, failed ports.Transport) { | 173 | func (d *Daemon) finishSendErrorDetach(sess *session, ac *attachedClient, failed ports.Transport) { | ||
| 173 | ac.clearSamePeerOffer() | 174 | ac.clearSamePeerOffer() | ||
| 175 | + | ac.clearParkedRoute() | |||
| 174 | name := sess.nameSnapshot() | 176 | name := sess.nameSnapshot() | ||
| 175 | if rc := sess.renderCoordinator(); rc != nil { | 177 | if rc := sess.renderCoordinator(); rc != nil { | ||
| 176 | rc.noteDetach(ac) | 178 | rc.noteDetach(ac) | ||
M internal/usecase/daemon/clipboard.go +3 -0
| @@ -268,6 +268,9 @@ func (d *Daemon) boundedSendClipboardForward(item clipboardForward, ticket *atta | |||
| 268 | 268 | if ticket.ended.Load() || !ac.transportSnapshotCurrent(expected) { | |
| 269 | 269 | return errAttachmentTransition | |
| 270 | 270 | } | |
| 271 | + | if ac.parkedRouteOutput.Load() || ac.parkedRouteFullPending.Load() { | |
| 272 | + | return nil | |
| 273 | + | } | |
| 271 | 274 | if !beginClipboardOwnerSend(item.owner, ticket, expected) { | |
| 272 | 275 | return errAttachmentTransition | |
| 273 | 276 | } | |
| @@ -268,6 +268,9 @@ func (d *Daemon) boundedSendClipboardForward(item clipboardForward, ticket *atta | |||||
| 268 | if ticket.ended.Load() || !ac.transportSnapshotCurrent(expected) { | 268 | if ticket.ended.Load() || !ac.transportSnapshotCurrent(expected) { | ||
| 269 | return errAttachmentTransition | 269 | return errAttachmentTransition | ||
| 270 | } | 270 | } | ||
| 271 | + | if ac.parkedRouteOutput.Load() || ac.parkedRouteFullPending.Load() { | |||
| 272 | + | return nil | |||
| 273 | + | } | |||
| 271 | if !beginClipboardOwnerSend(item.owner, ticket, expected) { | 274 | if !beginClipboardOwnerSend(item.owner, ticket, expected) { | ||
| 272 | return errAttachmentTransition | 275 | return errAttachmentTransition | ||
| 273 | } | 276 | } | ||
A internal/usecase/daemon/parked_route.go +478 -0
| @@ -0,0 +1,478 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "crypto/rand" | |
| 6 | + | "io" | |
| 7 | + | "time" | |
| 8 | + | ||
| 9 | + | "github.com/bnema/vev/internal/domain" | |
| 10 | + | "github.com/bnema/vev/internal/ports" | |
| 11 | + | ) | |
| 12 | + | ||
| 13 | + | const parkedRouteLeaseTTL = 15 * time.Minute | |
| 14 | + | ||
| 15 | + | type parkedRouteLease struct { | |
| 16 | + | id ports.ParkedRouteLeaseID | |
| 17 | + | generation uint64 | |
| 18 | + | transport transportSnapshot | |
| 19 | + | expiresAt time.Time | |
| 20 | + | timer ports.Timer | |
| 21 | + | stop chan struct{} | |
| 22 | + | suspended bool | |
| 23 | + | inFlight bool | |
| 24 | + | expired bool | |
| 25 | + | } | |
| 26 | + | ||
| 27 | + | func newParkedRouteLeaseID() (ports.ParkedRouteLeaseID, error) { | |
| 28 | + | var id ports.ParkedRouteLeaseID | |
| 29 | + | if _, err := io.ReadFull(rand.Reader, id[:]); err != nil { | |
| 30 | + | return ports.ParkedRouteLeaseID{}, err | |
| 31 | + | } | |
| 32 | + | return id, nil | |
| 33 | + | } | |
| 34 | + | ||
| 35 | + | func stopParkedRouteLeaseLocked(lease *parkedRouteLease) { | |
| 36 | + | if lease == nil { | |
| 37 | + | return | |
| 38 | + | } | |
| 39 | + | if lease.timer != nil { | |
| 40 | + | lease.timer.Stop() | |
| 41 | + | lease.timer = nil | |
| 42 | + | } | |
| 43 | + | if lease.stop != nil { | |
| 44 | + | close(lease.stop) | |
| 45 | + | lease.stop = nil | |
| 46 | + | } | |
| 47 | + | } | |
| 48 | + | ||
| 49 | + | func (d *Daemon) armParkedRoute(token attachmentConnectionToken) (ports.ParkedRouteLeaseID, error) { | |
| 50 | + | ac := token.ac | |
| 51 | + | if ac == nil || !token.attachmentCurrent() || ac.navigationCapabilities&ports.NavigationCapabilityHomePicker == 0 { | |
| 52 | + | return ports.ParkedRouteLeaseID{}, errAttachmentTransition | |
| 53 | + | } | |
| 54 | + | id, err := newParkedRouteLeaseID() | |
| 55 | + | if err != nil { | |
| 56 | + | return ports.ParkedRouteLeaseID{}, err | |
| 57 | + | } | |
| 58 | + | lease := &parkedRouteLease{ | |
| 59 | + | id: id, generation: token.generation, transport: token.transport, | |
| 60 | + | expiresAt: d.clock.Now().Add(parkedRouteLeaseTTL), | |
| 61 | + | timer: d.clock.NewTimer(parkedRouteLeaseTTL), stop: make(chan struct{}), | |
| 62 | + | } | |
| 63 | + | ac.parkedRouteMu.Lock() | |
| 64 | + | stopParkedRouteLeaseLocked(ac.parkedRoute) | |
| 65 | + | ac.parkedRoute = lease | |
| 66 | + | ac.parkedRouteOutput.Store(false) | |
| 67 | + | ac.parkedRouteFullPending.Store(false) | |
| 68 | + | timer, stop := lease.timer, lease.stop | |
| 69 | + | ac.parkedRouteMu.Unlock() | |
| 70 | + | go d.watchParkedRouteLease(ac, lease, timer, stop) | |
| 71 | + | return id, nil | |
| 72 | + | } | |
| 73 | + | ||
| 74 | + | func (d *Daemon) watchParkedRouteLease(ac *attachedClient, lease *parkedRouteLease, timer ports.Timer, stop <-chan struct{}) { | |
| 75 | + | select { | |
| 76 | + | case <-timer.C(): | |
| 77 | + | d.expireParkedRoute(ac, lease) | |
| 78 | + | case <-stop: | |
| 79 | + | } | |
| 80 | + | } | |
| 81 | + | ||
| 82 | + | func (d *Daemon) expireParkedRoute(ac *attachedClient, lease *parkedRouteLease) { | |
| 83 | + | if ac == nil || lease == nil { | |
| 84 | + | return | |
| 85 | + | } | |
| 86 | + | ac.parkedRouteMu.Lock() | |
| 87 | + | if ac.parkedRoute != lease { | |
| 88 | + | ac.parkedRouteMu.Unlock() | |
| 89 | + | return | |
| 90 | + | } | |
| 91 | + | lease.timer = nil | |
| 92 | + | lease.stop = nil | |
| 93 | + | if lease.inFlight { | |
| 94 | + | lease.expired = true | |
| 95 | + | ac.parkedRouteMu.Unlock() | |
| 96 | + | return | |
| 97 | + | } | |
| 98 | + | ac.parkedRoute = nil | |
| 99 | + | if !lease.suspended { | |
| 100 | + | ac.parkedRouteOutput.Store(false) | |
| 101 | + | } | |
| 102 | + | ac.parkedRouteFullPending.Store(false) | |
| 103 | + | ac.parkedRouteMu.Unlock() | |
| 104 | + | if ac.connectionGeneration.Load() == lease.generation && ac.transportSnapshotCurrent(lease.transport) { | |
| 105 | + | _ = ac.closeCapturedTransport(lease.transport.transport) | |
| 106 | + | } | |
| 107 | + | } | |
| 108 | + | ||
| 109 | + | func (ac *attachedClient) clearParkedRoute() { | |
| 110 | + | if ac == nil { | |
| 111 | + | return | |
| 112 | + | } | |
| 113 | + | ac.parkedRouteMu.Lock() | |
| 114 | + | stopParkedRouteLeaseLocked(ac.parkedRoute) | |
| 115 | + | ac.parkedRoute = nil | |
| 116 | + | ac.parkedRouteOutput.Store(false) | |
| 117 | + | ac.parkedRouteFullPending.Store(false) | |
| 118 | + | ac.parkedRouteMu.Unlock() | |
| 119 | + | } | |
| 120 | + | ||
| 121 | + | func (ac *attachedClient) parkedRouteStatus(token attachmentConnectionToken, id ports.ParkedRouteLeaseID, now time.Time, suspended bool) ports.ParkedRouteStatus { | |
| 122 | + | if ac == nil || !token.attachmentCurrent() { | |
| 123 | + | return ports.ParkedRouteRejected | |
| 124 | + | } | |
| 125 | + | ac.parkedRouteMu.Lock() | |
| 126 | + | defer ac.parkedRouteMu.Unlock() | |
| 127 | + | lease := ac.parkedRoute | |
| 128 | + | if lease == nil || lease.id != id || lease.generation != token.generation || lease.transport != token.transport { | |
| 129 | + | return ports.ParkedRouteRejected | |
| 130 | + | } | |
| 131 | + | if !now.Before(lease.expiresAt) { | |
| 132 | + | stopParkedRouteLeaseLocked(lease) | |
| 133 | + | ac.parkedRoute = nil | |
| 134 | + | ac.parkedRouteOutput.Store(false) | |
| 135 | + | ac.parkedRouteFullPending.Store(false) | |
| 136 | + | return ports.ParkedRouteExpired | |
| 137 | + | } | |
| 138 | + | if lease.suspended != suspended { | |
| 139 | + | return ports.ParkedRouteRejected | |
| 140 | + | } | |
| 141 | + | return 0 | |
| 142 | + | } | |
| 143 | + | ||
| 144 | + | func (ac *attachedClient) prepareParkedRoute(token attachmentConnectionToken, id ports.ParkedRouteLeaseID, now time.Time) ports.ParkedRouteStatus { | |
| 145 | + | if status := ac.parkedRouteStatus(token, id, now, false); status != 0 { | |
| 146 | + | return status | |
| 147 | + | } | |
| 148 | + | ac.parkedRouteMu.Lock() | |
| 149 | + | defer ac.parkedRouteMu.Unlock() | |
| 150 | + | lease := ac.parkedRoute | |
| 151 | + | if lease == nil || lease.id != id || lease.generation != token.generation || lease.transport != token.transport || lease.suspended { | |
| 152 | + | return ports.ParkedRouteRejected | |
| 153 | + | } | |
| 154 | + | lease.suspended = true | |
| 155 | + | ac.parkedRouteOutput.Store(true) | |
| 156 | + | return 0 | |
| 157 | + | } | |
| 158 | + | ||
| 159 | + | func (ac *attachedClient) consumeParkedRoute(token attachmentConnectionToken, id ports.ParkedRouteLeaseID, now time.Time) ports.ParkedRouteStatus { | |
| 160 | + | if status := ac.parkedRouteStatus(token, id, now, true); status != 0 { | |
| 161 | + | return status | |
| 162 | + | } | |
| 163 | + | ac.parkedRouteMu.Lock() | |
| 164 | + | defer ac.parkedRouteMu.Unlock() | |
| 165 | + | lease := ac.parkedRoute | |
| 166 | + | if lease == nil || lease.id != id || lease.generation != token.generation || lease.transport != token.transport || !lease.suspended { | |
| 167 | + | return ports.ParkedRouteRejected | |
| 168 | + | } | |
| 169 | + | stopParkedRouteLeaseLocked(lease) | |
| 170 | + | ac.parkedRoute = nil | |
| 171 | + | return 0 | |
| 172 | + | } | |
| 173 | + | ||
| 174 | + | func (ac *attachedClient) beginParkedRouteSwitch(token attachmentConnectionToken, id ports.ParkedRouteLeaseID, now time.Time) ports.ParkedRouteStatus { | |
| 175 | + | if ac == nil || !token.attachmentCurrent() { | |
| 176 | + | return ports.ParkedRouteRejected | |
| 177 | + | } | |
| 178 | + | ac.parkedRouteMu.Lock() | |
| 179 | + | defer ac.parkedRouteMu.Unlock() | |
| 180 | + | lease := ac.parkedRoute | |
| 181 | + | if lease == nil || lease.id != id || lease.generation != token.generation || lease.transport != token.transport || !lease.suspended || lease.inFlight { | |
| 182 | + | return ports.ParkedRouteRejected | |
| 183 | + | } | |
| 184 | + | if lease.expired || !now.Before(lease.expiresAt) { | |
| 185 | + | stopParkedRouteLeaseLocked(lease) | |
| 186 | + | ac.parkedRoute = nil | |
| 187 | + | return ports.ParkedRouteExpired | |
| 188 | + | } | |
| 189 | + | lease.inFlight = true | |
| 190 | + | return 0 | |
| 191 | + | } | |
| 192 | + | ||
| 193 | + | func (ac *attachedClient) rejectParkedRouteSwitch(token attachmentConnectionToken, id ports.ParkedRouteLeaseID, now time.Time) ports.ParkedRouteStatus { | |
| 194 | + | if ac == nil { | |
| 195 | + | return ports.ParkedRouteRejected | |
| 196 | + | } | |
| 197 | + | ac.parkedRouteMu.Lock() | |
| 198 | + | defer ac.parkedRouteMu.Unlock() | |
| 199 | + | lease := ac.parkedRoute | |
| 200 | + | if lease == nil || lease.id != id || lease.generation != token.generation || lease.transport != token.transport || !lease.inFlight { | |
| 201 | + | return ports.ParkedRouteRejected | |
| 202 | + | } | |
| 203 | + | lease.inFlight = false | |
| 204 | + | if lease.expired || !now.Before(lease.expiresAt) { | |
| 205 | + | stopParkedRouteLeaseLocked(lease) | |
| 206 | + | ac.parkedRoute = nil | |
| 207 | + | return ports.ParkedRouteExpired | |
| 208 | + | } | |
| 209 | + | return 0 | |
| 210 | + | } | |
| 211 | + | ||
| 212 | + | func (ac *attachedClient) consumePublishedParkedRoute(previous, published attachmentConnectionToken, id ports.ParkedRouteLeaseID) ports.ParkedRouteStatus { | |
| 213 | + | if ac == nil || published.ac != ac || !published.attachmentCurrent() { | |
| 214 | + | return ports.ParkedRouteRejected | |
| 215 | + | } | |
| 216 | + | ac.parkedRouteMu.Lock() | |
| 217 | + | defer ac.parkedRouteMu.Unlock() | |
| 218 | + | lease := ac.parkedRoute | |
| 219 | + | if lease == nil || lease.id != id || lease.generation != previous.generation || lease.transport != previous.transport || !lease.suspended || !lease.inFlight { | |
| 220 | + | return ports.ParkedRouteRejected | |
| 221 | + | } | |
| 222 | + | // Expiry is checked when the request is admitted. Once publication starts, | |
| 223 | + | // the accepted request must commit atomically instead of turning a completed | |
| 224 | + | // identity publication into a late pre-commit rejection. | |
| 225 | + | stopParkedRouteLeaseLocked(lease) | |
| 226 | + | ac.parkedRoute = nil | |
| 227 | + | return 0 | |
| 228 | + | } | |
| 229 | + | ||
| 230 | + | func parkedRouteResponseFrame(response ports.ParkedRouteResponse) (ports.Frame, error) { | |
| 231 | + | payload := ports.MarshalParkedRouteResponse(response) | |
| 232 | + | if payload == nil { | |
| 233 | + | return ports.Frame{}, errAttachmentTransition | |
| 234 | + | } | |
| 235 | + | return ports.Frame{Type: ports.MsgParkedRouteResponse, Payload: payload}, nil | |
| 236 | + | } | |
| 237 | + | ||
| 238 | + | func (d *Daemon) sendParkedRouteResponse(token attachmentConnectionToken, response ports.ParkedRouteResponse) error { | |
| 239 | + | frame, err := parkedRouteResponseFrame(response) | |
| 240 | + | if err != nil { | |
| 241 | + | return err | |
| 242 | + | } | |
| 243 | + | return token.sendControl(frame) | |
| 244 | + | } | |
| 245 | + | ||
| 246 | + | // sendParkedRouteResponseLocked preserves response -> full-output ordering. | |
| 247 | + | // Caller holds ac.sendMu and keeps parkedRouteOutput set through this send. | |
| 248 | + | func (d *Daemon) sendParkedRouteResponseLocked(ac *attachedClient, expected transportSnapshot, response ports.ParkedRouteResponse) error { | |
| 249 | + | frame, err := parkedRouteResponseFrame(response) | |
| 250 | + | if err != nil { | |
| 251 | + | return err | |
| 252 | + | } | |
| 253 | + | if ac == nil || !ac.transportSnapshotCurrent(expected) || expected.transport == nil { | |
| 254 | + | return errAttachmentTransition | |
| 255 | + | } | |
| 256 | + | return expected.transport.Send(frame) | |
| 257 | + | } | |
| 258 | + | ||
| 259 | + | func (ac *attachedClient) releaseParkedOutputLocked() { | |
| 260 | + | ac.rebaseOutput() | |
| 261 | + | ac.pipelineCache = composeCacheInput{} | |
| 262 | + | ac.pipelineScratch = composeCacheInput{} | |
| 263 | + | ac.parkedRouteFullPending.Store(true) | |
| 264 | + | ac.parkedRouteOutput.Store(false) | |
| 265 | + | } | |
| 266 | + | ||
| 267 | + | func (d *Daemon) parkedRouteResponseFailed(token attachmentConnectionToken) { | |
| 268 | + | if token.effect == nil || token.effect.ended.Load() { | |
| 269 | + | return | |
| 270 | + | } | |
| 271 | + | launch := d.reserveAttachmentSendErrorCleanup(token, token.transport.transport) | |
| 272 | + | token.effect.End() | |
| 273 | + | launch() | |
| 274 | + | } | |
| 275 | + | ||
| 276 | + | func (d *Daemon) respondParkedRoute(token attachmentConnectionToken, response ports.ParkedRouteResponse) bool { | |
| 277 | + | if err := d.sendParkedRouteResponse(token, response); err != nil { | |
| 278 | + | d.parkedRouteResponseFailed(token) | |
| 279 | + | return false | |
| 280 | + | } | |
| 281 | + | return true | |
| 282 | + | } | |
| 283 | + | ||
| 284 | + | func (d *Daemon) handleParkedRouteRequest(token attachmentConnectionToken, request ports.ParkedRouteRequest) { | |
| 285 | + | if ports.ValidateParkedRouteRequest(request) != nil || token.ac == nil { | |
| 286 | + | return | |
| 287 | + | } | |
| 288 | + | switch request.Action { | |
| 289 | + | case ports.ParkedRoutePrepare: | |
| 290 | + | status := token.ac.prepareParkedRoute(token, request.LeaseID, d.clock.Now()) | |
| 291 | + | if status == 0 { | |
| 292 | + | status = ports.ParkedRouteReady | |
| 293 | + | } | |
| 294 | + | d.respondParkedRoute(token, ports.ParkedRouteResponse{RequestID: request.RequestID, Status: status}) | |
| 295 | + | case ports.ParkedRouteResume: | |
| 296 | + | d.resumeParkedRoute(token, request) | |
| 297 | + | case ports.ParkedRouteSwitch: | |
| 298 | + | d.switchParkedRoute(token, request) | |
| 299 | + | } | |
| 300 | + | } | |
| 301 | + | ||
| 302 | + | func (d *Daemon) resumeParkedRoute(token attachmentConnectionToken, request ports.ParkedRouteRequest) { | |
| 303 | + | status := token.ac.consumeParkedRoute(token, request.LeaseID, d.clock.Now()) | |
| 304 | + | if status != 0 { | |
| 305 | + | d.respondParkedRoute(token, ports.ParkedRouteResponse{RequestID: request.RequestID, Status: status}) | |
| 306 | + | return | |
| 307 | + | } | |
| 308 | + | ac := token.ac | |
| 309 | + | ac.sendMu.Lock() | |
| 310 | + | err := d.sendParkedRouteResponseLocked(ac, token.transport, ports.ParkedRouteResponse{RequestID: request.RequestID, Status: ports.ParkedRouteResumed}) | |
| 311 | + | if err == nil { | |
| 312 | + | ac.releaseParkedOutputLocked() | |
| 313 | + | } | |
| 314 | + | ac.sendMu.Unlock() | |
| 315 | + | if err != nil { | |
| 316 | + | d.parkedRouteResponseFailed(token) | |
| 317 | + | return | |
| 318 | + | } | |
| 319 | + | if token.sess != nil { | |
| 320 | + | d.firstPaintWithLease(token.sess, ac, token.lease, true) | |
| 321 | + | } | |
| 322 | + | } | |
| 323 | + | ||
| 324 | + | func (d *Daemon) rejectParkedRouteSwitch(token attachmentConnectionToken, request ports.ParkedRouteRequest, fallback ports.ParkedRouteStatus) { | |
| 325 | + | status := token.ac.rejectParkedRouteSwitch(token, request.LeaseID, d.clock.Now()) | |
| 326 | + | if status == 0 { | |
| 327 | + | status = fallback | |
| 328 | + | } | |
| 329 | + | if d.respondParkedRoute(token, ports.ParkedRouteResponse{RequestID: request.RequestID, Status: status}) && status == ports.ParkedRouteExpired { | |
| 330 | + | _ = token.ac.closeCapturedTransport(token.transport.transport) | |
| 331 | + | } | |
| 332 | + | } | |
| 333 | + | ||
| 334 | + | func (d *Daemon) switchParkedRoute(token attachmentConnectionToken, request ports.ParkedRouteRequest) { | |
| 335 | + | if status := token.ac.beginParkedRouteSwitch(token, request.LeaseID, d.clock.Now()); status != 0 { | |
| 336 | + | if d.respondParkedRoute(token, ports.ParkedRouteResponse{RequestID: request.RequestID, Status: status}) && status == ports.ParkedRouteExpired { | |
| 337 | + | _ = token.ac.closeCapturedTransport(token.transport.transport) | |
| 338 | + | } | |
| 339 | + | return | |
| 340 | + | } | |
| 341 | + | target := *request.Target | |
| 342 | + | ctx := d.serveCtx | |
| 343 | + | if ctx == nil { | |
| 344 | + | ctx = context.Background() | |
| 345 | + | } | |
| 346 | + | if err := d.waitForParkedTargetRestore(ctx, target.SessionName); err != nil { | |
| 347 | + | d.rejectParkedRouteSwitch(token, request, ports.ParkedRouteStaleTarget) | |
| 348 | + | return | |
| 349 | + | } | |
| 350 | + | ||
| 351 | + | transition, targetSession, err := d.transitionParkedRouteTarget(token, target) | |
| 352 | + | if err != nil { | |
| 353 | + | d.rejectParkedRouteSwitch(token, request, ports.ParkedRouteStaleTarget) | |
| 354 | + | return | |
| 355 | + | } | |
| 356 | + | if status := token.ac.consumePublishedParkedRoute(token, transition.published, request.LeaseID); status != 0 { | |
| 357 | + | d.abortPublishedAttachmentTransition(transition) | |
| 358 | + | return | |
| 359 | + | } | |
| 360 | + | fresh, admitted := token.ac.beginAttachmentEffect(transition.published) | |
| 361 | + | if !admitted { | |
| 362 | + | d.abortPublishedAttachmentTransition(transition) | |
| 363 | + | return | |
| 364 | + | } | |
| 365 | + | freshToken := fresh.connectionToken() | |
| 366 | + | ac := token.ac | |
| 367 | + | ac.sendMu.Lock() | |
| 368 | + | responseErr := d.sendParkedRouteResponseLocked(ac, freshToken.transport, ports.ParkedRouteResponse{RequestID: request.RequestID, Status: ports.ParkedRouteSwitched}) | |
| 369 | + | if responseErr == nil { | |
| 370 | + | ac.releaseParkedOutputLocked() | |
| 371 | + | } | |
| 372 | + | ac.sendMu.Unlock() | |
| 373 | + | if responseErr != nil { | |
| 374 | + | launch := d.reserveAttachmentSendErrorCleanup(freshToken, freshToken.transport.transport) | |
| 375 | + | fresh.End() | |
| 376 | + | launch() | |
| 377 | + | d.deferAttachmentTransitionCleanups(transition) | |
| 378 | + | return | |
| 379 | + | } | |
| 380 | + | fresh.End() | |
| 381 | + | d.touchMRU(targetSession) | |
| 382 | + | d.deferAttachmentTransitionCleanups(transition) | |
| 383 | + | transition.published.rebase = false | |
| 384 | + | d.firstPaintForTransition(transition.published) | |
| 385 | + | } | |
| 386 | + | ||
| 387 | + | func (d *Daemon) waitForParkedTargetRestore(parent context.Context, name string) error { | |
| 388 | + | ctx, cancel := context.WithCancel(parent) | |
| 389 | + | timer := d.clock.NewTimer(ports.HandshakeTimeout) | |
| 390 | + | stop := make(chan struct{}) | |
| 391 | + | go func() { | |
| 392 | + | select { | |
| 393 | + | case <-timer.C(): | |
| 394 | + | cancel() | |
| 395 | + | case <-ctx.Done(): | |
| 396 | + | case <-stop: | |
| 397 | + | } | |
| 398 | + | }() | |
| 399 | + | err := d.waitForTargetRestore(ctx, name) | |
| 400 | + | close(stop) | |
| 401 | + | timer.Stop() | |
| 402 | + | cancel() | |
| 403 | + | return err | |
| 404 | + | } | |
| 405 | + | ||
| 406 | + | func (d *Daemon) transitionParkedRouteTarget(token attachmentConnectionToken, target domain.RemoteSessionTarget) (attachmentTransitionResult, *session, error) { | |
| 407 | + | d.mu.Lock() | |
| 408 | + | if d.closing { | |
| 409 | + | d.mu.Unlock() | |
| 410 | + | return attachmentTransitionResult{}, nil, errAttachmentTransition | |
| 411 | + | } | |
| 412 | + | if live := d.findByNameLocked(target.SessionName); live != nil { | |
| 413 | + | if live.incarnation != target.LifecycleID { | |
| 414 | + | d.mu.Unlock() | |
| 415 | + | return attachmentTransitionResult{}, nil, errAttachmentTransition | |
| 416 | + | } | |
| 417 | + | tabIndex, ok := remoteTargetTabIndexLocked(live, target) | |
| 418 | + | if !ok { | |
| 419 | + | d.mu.Unlock() | |
| 420 | + | return attachmentTransitionResult{}, nil, errAttachmentTransition | |
| 421 | + | } | |
| 422 | + | live.mu.Lock() | |
| 423 | + | tabID := domain.TabStableID(live.tabs[tabIndex].stableID) | |
| 424 | + | live.mu.Unlock() | |
| 425 | + | d.mu.Unlock() | |
| 426 | + | transition, err := d.transitionAttachment(attachmentTransitionRequest{ | |
| 427 | + | source: token.sess, target: live, next: token.ac, | |
| 428 | + | expectedTransport: token.transport, sourceToken: &token, action: "parked-route-switch", | |
| 429 | + | expectedTargetLifecycle: &attachmentLifecycleFence{ | |
| 430 | + | name: target.SessionName, incarnation: target.LifecycleID, checkIncarnation: true, | |
| 431 | + | tabID: tabID, tabIndex: tabIndex, checkTab: true, | |
| 432 | + | }, | |
| 433 | + | activateTargetTab: true, targetTabIndex: tabIndex, ready: true, | |
| 434 | + | }) | |
| 435 | + | return transition, live, err | |
| 436 | + | } | |
| 437 | + | ||
| 438 | + | inactive, ok := d.inactive[target.SessionName] | |
| 439 | + | if !ok || !target.Stopped || inactive.incarnation != target.LifecycleID || !inactive.canResume() { | |
| 440 | + | d.mu.Unlock() | |
| 441 | + | return attachmentTransitionResult{}, nil, errAttachmentTransition | |
| 442 | + | } | |
| 443 | + | tabIndex, ok := remoteTargetTabIndexInactive(inactive, target) | |
| 444 | + | if !ok { | |
| 445 | + | d.mu.Unlock() | |
| 446 | + | return attachmentTransitionResult{}, nil, errAttachmentTransition | |
| 447 | + | } | |
| 448 | + | d.mu.Unlock() | |
| 449 | + | ||
| 450 | + | var created *session | |
| 451 | + | transition, err := d.transitionAttachment(attachmentTransitionRequest{ | |
| 452 | + | source: token.sess, next: token.ac, | |
| 453 | + | expectedTransport: token.transport, sourceToken: &token, action: "parked-route-restore", | |
| 454 | + | activateTargetTab: true, targetTabIndex: tabIndex, ready: true, | |
| 455 | + | createTargetLocked: func() (*session, error) { | |
| 456 | + | current, ok := d.inactive[target.SessionName] | |
| 457 | + | if !ok || current.incarnation != target.LifecycleID || !current.canResume() { | |
| 458 | + | return nil, errAttachmentTransition | |
| 459 | + | } | |
| 460 | + | resolved, ok := remoteTargetTabIndexInactive(current, target) | |
| 461 | + | if !ok || resolved != tabIndex { | |
| 462 | + | return nil, errAttachmentTransition | |
| 463 | + | } | |
| 464 | + | env := copyEnvironment(d.baseEnv) | |
| 465 | + | cwd := d.dirOrHome(current.cwd) | |
| 466 | + | var createErr error | |
| 467 | + | created, createErr = d.resumeRemoteInactiveSessionLocked(target, cwd, token.ac.sizeSnapshot(), env, current) | |
| 468 | + | return created, createErr | |
| 469 | + | }, | |
| 470 | + | }) | |
| 471 | + | if err != nil && created != nil { | |
| 472 | + | _ = d.killSession(created, ports.ReasonSessionKilled, true) | |
| 473 | + | } | |
| 474 | + | if err != nil { | |
| 475 | + | return attachmentTransitionResult{}, created, err | |
| 476 | + | } | |
| 477 | + | return transition, created, nil | |
| 478 | + | } | |
| @@ -0,0 +1,478 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "crypto/rand" | |||
| 6 | + | "io" | |||
| 7 | + | "time" | |||
| 8 | + | ||||
| 9 | + | "github.com/bnema/vev/internal/domain" | |||
| 10 | + | "github.com/bnema/vev/internal/ports" | |||
| 11 | + | ) | |||
| 12 | + | ||||
| 13 | + | const parkedRouteLeaseTTL = 15 * time.Minute | |||
| 14 | + | ||||
| 15 | + | type parkedRouteLease struct { | |||
| 16 | + | id ports.ParkedRouteLeaseID | |||
| 17 | + | generation uint64 | |||
| 18 | + | transport transportSnapshot | |||
| 19 | + | expiresAt time.Time | |||
| 20 | + | timer ports.Timer | |||
| 21 | + | stop chan struct{} | |||
| 22 | + | suspended bool | |||
| 23 | + | inFlight bool | |||
| 24 | + | expired bool | |||
| 25 | + | } | |||
| 26 | + | ||||
| 27 | + | func newParkedRouteLeaseID() (ports.ParkedRouteLeaseID, error) { | |||
| 28 | + | var id ports.ParkedRouteLeaseID | |||
| 29 | + | if _, err := io.ReadFull(rand.Reader, id[:]); err != nil { | |||
| 30 | + | return ports.ParkedRouteLeaseID{}, err | |||
| 31 | + | } | |||
| 32 | + | return id, nil | |||
| 33 | + | } | |||
| 34 | + | ||||
| 35 | + | func stopParkedRouteLeaseLocked(lease *parkedRouteLease) { | |||
| 36 | + | if lease == nil { | |||
| 37 | + | return | |||
| 38 | + | } | |||
| 39 | + | if lease.timer != nil { | |||
| 40 | + | lease.timer.Stop() | |||
| 41 | + | lease.timer = nil | |||
| 42 | + | } | |||
| 43 | + | if lease.stop != nil { | |||
| 44 | + | close(lease.stop) | |||
| 45 | + | lease.stop = nil | |||
| 46 | + | } | |||
| 47 | + | } | |||
| 48 | + | ||||
| 49 | + | func (d *Daemon) armParkedRoute(token attachmentConnectionToken) (ports.ParkedRouteLeaseID, error) { | |||
| 50 | + | ac := token.ac | |||
| 51 | + | if ac == nil || !token.attachmentCurrent() || ac.navigationCapabilities&ports.NavigationCapabilityHomePicker == 0 { | |||
| 52 | + | return ports.ParkedRouteLeaseID{}, errAttachmentTransition | |||
| 53 | + | } | |||
| 54 | + | id, err := newParkedRouteLeaseID() | |||
| 55 | + | if err != nil { | |||
| 56 | + | return ports.ParkedRouteLeaseID{}, err | |||
| 57 | + | } | |||
| 58 | + | lease := &parkedRouteLease{ | |||
| 59 | + | id: id, generation: token.generation, transport: token.transport, | |||
| 60 | + | expiresAt: d.clock.Now().Add(parkedRouteLeaseTTL), | |||
| 61 | + | timer: d.clock.NewTimer(parkedRouteLeaseTTL), stop: make(chan struct{}), | |||
| 62 | + | } | |||
| 63 | + | ac.parkedRouteMu.Lock() | |||
| 64 | + | stopParkedRouteLeaseLocked(ac.parkedRoute) | |||
| 65 | + | ac.parkedRoute = lease | |||
| 66 | + | ac.parkedRouteOutput.Store(false) | |||
| 67 | + | ac.parkedRouteFullPending.Store(false) | |||
| 68 | + | timer, stop := lease.timer, lease.stop | |||
| 69 | + | ac.parkedRouteMu.Unlock() | |||
| 70 | + | go d.watchParkedRouteLease(ac, lease, timer, stop) | |||
| 71 | + | return id, nil | |||
| 72 | + | } | |||
| 73 | + | ||||
| 74 | + | func (d *Daemon) watchParkedRouteLease(ac *attachedClient, lease *parkedRouteLease, timer ports.Timer, stop <-chan struct{}) { | |||
| 75 | + | select { | |||
| 76 | + | case <-timer.C(): | |||
| 77 | + | d.expireParkedRoute(ac, lease) | |||
| 78 | + | case <-stop: | |||
| 79 | + | } | |||
| 80 | + | } | |||
| 81 | + | ||||
| 82 | + | func (d *Daemon) expireParkedRoute(ac *attachedClient, lease *parkedRouteLease) { | |||
| 83 | + | if ac == nil || lease == nil { | |||
| 84 | + | return | |||
| 85 | + | } | |||
| 86 | + | ac.parkedRouteMu.Lock() | |||
| 87 | + | if ac.parkedRoute != lease { | |||
| 88 | + | ac.parkedRouteMu.Unlock() | |||
| 89 | + | return | |||
| 90 | + | } | |||
| 91 | + | lease.timer = nil | |||
| 92 | + | lease.stop = nil | |||
| 93 | + | if lease.inFlight { | |||
| 94 | + | lease.expired = true | |||
| 95 | + | ac.parkedRouteMu.Unlock() | |||
| 96 | + | return | |||
| 97 | + | } | |||
| 98 | + | ac.parkedRoute = nil | |||
| 99 | + | if !lease.suspended { | |||
| 100 | + | ac.parkedRouteOutput.Store(false) | |||
| 101 | + | } | |||
| 102 | + | ac.parkedRouteFullPending.Store(false) | |||
| 103 | + | ac.parkedRouteMu.Unlock() | |||
| 104 | + | if ac.connectionGeneration.Load() == lease.generation && ac.transportSnapshotCurrent(lease.transport) { | |||
| 105 | + | _ = ac.closeCapturedTransport(lease.transport.transport) | |||
| 106 | + | } | |||
| 107 | + | } | |||
| 108 | + | ||||
| 109 | + | func (ac *attachedClient) clearParkedRoute() { | |||
| 110 | + | if ac == nil { | |||
| 111 | + | return | |||
| 112 | + | } | |||
| 113 | + | ac.parkedRouteMu.Lock() | |||
| 114 | + | stopParkedRouteLeaseLocked(ac.parkedRoute) | |||
| 115 | + | ac.parkedRoute = nil | |||
| 116 | + | ac.parkedRouteOutput.Store(false) | |||
| 117 | + | ac.parkedRouteFullPending.Store(false) | |||
| 118 | + | ac.parkedRouteMu.Unlock() | |||
| 119 | + | } | |||
| 120 | + | ||||
| 121 | + | func (ac *attachedClient) parkedRouteStatus(token attachmentConnectionToken, id ports.ParkedRouteLeaseID, now time.Time, suspended bool) ports.ParkedRouteStatus { | |||
| 122 | + | if ac == nil || !token.attachmentCurrent() { | |||
| 123 | + | return ports.ParkedRouteRejected | |||
| 124 | + | } | |||
| 125 | + | ac.parkedRouteMu.Lock() | |||
| 126 | + | defer ac.parkedRouteMu.Unlock() | |||
| 127 | + | lease := ac.parkedRoute | |||
| 128 | + | if lease == nil || lease.id != id || lease.generation != token.generation || lease.transport != token.transport { | |||
| 129 | + | return ports.ParkedRouteRejected | |||
| 130 | + | } | |||
| 131 | + | if !now.Before(lease.expiresAt) { | |||
| 132 | + | stopParkedRouteLeaseLocked(lease) | |||
| 133 | + | ac.parkedRoute = nil | |||
| 134 | + | ac.parkedRouteOutput.Store(false) | |||
| 135 | + | ac.parkedRouteFullPending.Store(false) | |||
| 136 | + | return ports.ParkedRouteExpired | |||
| 137 | + | } | |||
| 138 | + | if lease.suspended != suspended { | |||
| 139 | + | return ports.ParkedRouteRejected | |||
| 140 | + | } | |||
| 141 | + | return 0 | |||
| 142 | + | } | |||
| 143 | + | ||||
| 144 | + | func (ac *attachedClient) prepareParkedRoute(token attachmentConnectionToken, id ports.ParkedRouteLeaseID, now time.Time) ports.ParkedRouteStatus { | |||
| 145 | + | if status := ac.parkedRouteStatus(token, id, now, false); status != 0 { | |||
| 146 | + | return status | |||
| 147 | + | } | |||
| 148 | + | ac.parkedRouteMu.Lock() | |||
| 149 | + | defer ac.parkedRouteMu.Unlock() | |||
| 150 | + | lease := ac.parkedRoute | |||
| 151 | + | if lease == nil || lease.id != id || lease.generation != token.generation || lease.transport != token.transport || lease.suspended { | |||
| 152 | + | return ports.ParkedRouteRejected | |||
| 153 | + | } | |||
| 154 | + | lease.suspended = true | |||
| 155 | + | ac.parkedRouteOutput.Store(true) | |||
| 156 | + | return 0 | |||
| 157 | + | } | |||
| 158 | + | ||||
| 159 | + | func (ac *attachedClient) consumeParkedRoute(token attachmentConnectionToken, id ports.ParkedRouteLeaseID, now time.Time) ports.ParkedRouteStatus { | |||
| 160 | + | if status := ac.parkedRouteStatus(token, id, now, true); status != 0 { | |||
| 161 | + | return status | |||
| 162 | + | } | |||
| 163 | + | ac.parkedRouteMu.Lock() | |||
| 164 | + | defer ac.parkedRouteMu.Unlock() | |||
| 165 | + | lease := ac.parkedRoute | |||
| 166 | + | if lease == nil || lease.id != id || lease.generation != token.generation || lease.transport != token.transport || !lease.suspended { | |||
| 167 | + | return ports.ParkedRouteRejected | |||
| 168 | + | } | |||
| 169 | + | stopParkedRouteLeaseLocked(lease) | |||
| 170 | + | ac.parkedRoute = nil | |||
| 171 | + | return 0 | |||
| 172 | + | } | |||
| 173 | + | ||||
| 174 | + | func (ac *attachedClient) beginParkedRouteSwitch(token attachmentConnectionToken, id ports.ParkedRouteLeaseID, now time.Time) ports.ParkedRouteStatus { | |||
| 175 | + | if ac == nil || !token.attachmentCurrent() { | |||
| 176 | + | return ports.ParkedRouteRejected | |||
| 177 | + | } | |||
| 178 | + | ac.parkedRouteMu.Lock() | |||
| 179 | + | defer ac.parkedRouteMu.Unlock() | |||
| 180 | + | lease := ac.parkedRoute | |||
| 181 | + | if lease == nil || lease.id != id || lease.generation != token.generation || lease.transport != token.transport || !lease.suspended || lease.inFlight { | |||
| 182 | + | return ports.ParkedRouteRejected | |||
| 183 | + | } | |||
| 184 | + | if lease.expired || !now.Before(lease.expiresAt) { | |||
| 185 | + | stopParkedRouteLeaseLocked(lease) | |||
| 186 | + | ac.parkedRoute = nil | |||
| 187 | + | return ports.ParkedRouteExpired | |||
| 188 | + | } | |||
| 189 | + | lease.inFlight = true | |||
| 190 | + | return 0 | |||
| 191 | + | } | |||
| 192 | + | ||||
| 193 | + | func (ac *attachedClient) rejectParkedRouteSwitch(token attachmentConnectionToken, id ports.ParkedRouteLeaseID, now time.Time) ports.ParkedRouteStatus { | |||
| 194 | + | if ac == nil { | |||
| 195 | + | return ports.ParkedRouteRejected | |||
| 196 | + | } | |||
| 197 | + | ac.parkedRouteMu.Lock() | |||
| 198 | + | defer ac.parkedRouteMu.Unlock() | |||
| 199 | + | lease := ac.parkedRoute | |||
| 200 | + | if lease == nil || lease.id != id || lease.generation != token.generation || lease.transport != token.transport || !lease.inFlight { | |||
| 201 | + | return ports.ParkedRouteRejected | |||
| 202 | + | } | |||
| 203 | + | lease.inFlight = false | |||
| 204 | + | if lease.expired || !now.Before(lease.expiresAt) { | |||
| 205 | + | stopParkedRouteLeaseLocked(lease) | |||
| 206 | + | ac.parkedRoute = nil | |||
| 207 | + | return ports.ParkedRouteExpired | |||
| 208 | + | } | |||
| 209 | + | return 0 | |||
| 210 | + | } | |||
| 211 | + | ||||
| 212 | + | func (ac *attachedClient) consumePublishedParkedRoute(previous, published attachmentConnectionToken, id ports.ParkedRouteLeaseID) ports.ParkedRouteStatus { | |||
| 213 | + | if ac == nil || published.ac != ac || !published.attachmentCurrent() { | |||
| 214 | + | return ports.ParkedRouteRejected | |||
| 215 | + | } | |||
| 216 | + | ac.parkedRouteMu.Lock() | |||
| 217 | + | defer ac.parkedRouteMu.Unlock() | |||
| 218 | + | lease := ac.parkedRoute | |||
| 219 | + | if lease == nil || lease.id != id || lease.generation != previous.generation || lease.transport != previous.transport || !lease.suspended || !lease.inFlight { | |||
| 220 | + | return ports.ParkedRouteRejected | |||
| 221 | + | } | |||
| 222 | + | // Expiry is checked when the request is admitted. Once publication starts, | |||
| 223 | + | // the accepted request must commit atomically instead of turning a completed | |||
| 224 | + | // identity publication into a late pre-commit rejection. | |||
| 225 | + | stopParkedRouteLeaseLocked(lease) | |||
| 226 | + | ac.parkedRoute = nil | |||
| 227 | + | return 0 | |||
| 228 | + | } | |||
| 229 | + | ||||
| 230 | + | func parkedRouteResponseFrame(response ports.ParkedRouteResponse) (ports.Frame, error) { | |||
| 231 | + | payload := ports.MarshalParkedRouteResponse(response) | |||
| 232 | + | if payload == nil { | |||
| 233 | + | return ports.Frame{}, errAttachmentTransition | |||
| 234 | + | } | |||
| 235 | + | return ports.Frame{Type: ports.MsgParkedRouteResponse, Payload: payload}, nil | |||
| 236 | + | } | |||
| 237 | + | ||||
| 238 | + | func (d *Daemon) sendParkedRouteResponse(token attachmentConnectionToken, response ports.ParkedRouteResponse) error { | |||
| 239 | + | frame, err := parkedRouteResponseFrame(response) | |||
| 240 | + | if err != nil { | |||
| 241 | + | return err | |||
| 242 | + | } | |||
| 243 | + | return token.sendControl(frame) | |||
| 244 | + | } | |||
| 245 | + | ||||
| 246 | + | // sendParkedRouteResponseLocked preserves response -> full-output ordering. | |||
| 247 | + | // Caller holds ac.sendMu and keeps parkedRouteOutput set through this send. | |||
| 248 | + | func (d *Daemon) sendParkedRouteResponseLocked(ac *attachedClient, expected transportSnapshot, response ports.ParkedRouteResponse) error { | |||
| 249 | + | frame, err := parkedRouteResponseFrame(response) | |||
| 250 | + | if err != nil { | |||
| 251 | + | return err | |||
| 252 | + | } | |||
| 253 | + | if ac == nil || !ac.transportSnapshotCurrent(expected) || expected.transport == nil { | |||
| 254 | + | return errAttachmentTransition | |||
| 255 | + | } | |||
| 256 | + | return expected.transport.Send(frame) | |||
| 257 | + | } | |||
| 258 | + | ||||
| 259 | + | func (ac *attachedClient) releaseParkedOutputLocked() { | |||
| 260 | + | ac.rebaseOutput() | |||
| 261 | + | ac.pipelineCache = composeCacheInput{} | |||
| 262 | + | ac.pipelineScratch = composeCacheInput{} | |||
| 263 | + | ac.parkedRouteFullPending.Store(true) | |||
| 264 | + | ac.parkedRouteOutput.Store(false) | |||
| 265 | + | } | |||
| 266 | + | ||||
| 267 | + | func (d *Daemon) parkedRouteResponseFailed(token attachmentConnectionToken) { | |||
| 268 | + | if token.effect == nil || token.effect.ended.Load() { | |||
| 269 | + | return | |||
| 270 | + | } | |||
| 271 | + | launch := d.reserveAttachmentSendErrorCleanup(token, token.transport.transport) | |||
| 272 | + | token.effect.End() | |||
| 273 | + | launch() | |||
| 274 | + | } | |||
| 275 | + | ||||
| 276 | + | func (d *Daemon) respondParkedRoute(token attachmentConnectionToken, response ports.ParkedRouteResponse) bool { | |||
| 277 | + | if err := d.sendParkedRouteResponse(token, response); err != nil { | |||
| 278 | + | d.parkedRouteResponseFailed(token) | |||
| 279 | + | return false | |||
| 280 | + | } | |||
| 281 | + | return true | |||
| 282 | + | } | |||
| 283 | + | ||||
| 284 | + | func (d *Daemon) handleParkedRouteRequest(token attachmentConnectionToken, request ports.ParkedRouteRequest) { | |||
| 285 | + | if ports.ValidateParkedRouteRequest(request) != nil || token.ac == nil { | |||
| 286 | + | return | |||
| 287 | + | } | |||
| 288 | + | switch request.Action { | |||
| 289 | + | case ports.ParkedRoutePrepare: | |||
| 290 | + | status := token.ac.prepareParkedRoute(token, request.LeaseID, d.clock.Now()) | |||
| 291 | + | if status == 0 { | |||
| 292 | + | status = ports.ParkedRouteReady | |||
| 293 | + | } | |||
| 294 | + | d.respondParkedRoute(token, ports.ParkedRouteResponse{RequestID: request.RequestID, Status: status}) | |||
| 295 | + | case ports.ParkedRouteResume: | |||
| 296 | + | d.resumeParkedRoute(token, request) | |||
| 297 | + | case ports.ParkedRouteSwitch: | |||
| 298 | + | d.switchParkedRoute(token, request) | |||
| 299 | + | } | |||
| 300 | + | } | |||
| 301 | + | ||||
| 302 | + | func (d *Daemon) resumeParkedRoute(token attachmentConnectionToken, request ports.ParkedRouteRequest) { | |||
| 303 | + | status := token.ac.consumeParkedRoute(token, request.LeaseID, d.clock.Now()) | |||
| 304 | + | if status != 0 { | |||
| 305 | + | d.respondParkedRoute(token, ports.ParkedRouteResponse{RequestID: request.RequestID, Status: status}) | |||
| 306 | + | return | |||
| 307 | + | } | |||
| 308 | + | ac := token.ac | |||
| 309 | + | ac.sendMu.Lock() | |||
| 310 | + | err := d.sendParkedRouteResponseLocked(ac, token.transport, ports.ParkedRouteResponse{RequestID: request.RequestID, Status: ports.ParkedRouteResumed}) | |||
| 311 | + | if err == nil { | |||
| 312 | + | ac.releaseParkedOutputLocked() | |||
| 313 | + | } | |||
| 314 | + | ac.sendMu.Unlock() | |||
| 315 | + | if err != nil { | |||
| 316 | + | d.parkedRouteResponseFailed(token) | |||
| 317 | + | return | |||
| 318 | + | } | |||
| 319 | + | if token.sess != nil { | |||
| 320 | + | d.firstPaintWithLease(token.sess, ac, token.lease, true) | |||
| 321 | + | } | |||
| 322 | + | } | |||
| 323 | + | ||||
| 324 | + | func (d *Daemon) rejectParkedRouteSwitch(token attachmentConnectionToken, request ports.ParkedRouteRequest, fallback ports.ParkedRouteStatus) { | |||
| 325 | + | status := token.ac.rejectParkedRouteSwitch(token, request.LeaseID, d.clock.Now()) | |||
| 326 | + | if status == 0 { | |||
| 327 | + | status = fallback | |||
| 328 | + | } | |||
| 329 | + | if d.respondParkedRoute(token, ports.ParkedRouteResponse{RequestID: request.RequestID, Status: status}) && status == ports.ParkedRouteExpired { | |||
| 330 | + | _ = token.ac.closeCapturedTransport(token.transport.transport) | |||
| 331 | + | } | |||
| 332 | + | } | |||
| 333 | + | ||||
| 334 | + | func (d *Daemon) switchParkedRoute(token attachmentConnectionToken, request ports.ParkedRouteRequest) { | |||
| 335 | + | if status := token.ac.beginParkedRouteSwitch(token, request.LeaseID, d.clock.Now()); status != 0 { | |||
| 336 | + | if d.respondParkedRoute(token, ports.ParkedRouteResponse{RequestID: request.RequestID, Status: status}) && status == ports.ParkedRouteExpired { | |||
| 337 | + | _ = token.ac.closeCapturedTransport(token.transport.transport) | |||
| 338 | + | } | |||
| 339 | + | return | |||
| 340 | + | } | |||
| 341 | + | target := *request.Target | |||
| 342 | + | ctx := d.serveCtx | |||
| 343 | + | if ctx == nil { | |||
| 344 | + | ctx = context.Background() | |||
| 345 | + | } | |||
| 346 | + | if err := d.waitForParkedTargetRestore(ctx, target.SessionName); err != nil { | |||
| 347 | + | d.rejectParkedRouteSwitch(token, request, ports.ParkedRouteStaleTarget) | |||
| 348 | + | return | |||
| 349 | + | } | |||
| 350 | + | ||||
| 351 | + | transition, targetSession, err := d.transitionParkedRouteTarget(token, target) | |||
| 352 | + | if err != nil { | |||
| 353 | + | d.rejectParkedRouteSwitch(token, request, ports.ParkedRouteStaleTarget) | |||
| 354 | + | return | |||
| 355 | + | } | |||
| 356 | + | if status := token.ac.consumePublishedParkedRoute(token, transition.published, request.LeaseID); status != 0 { | |||
| 357 | + | d.abortPublishedAttachmentTransition(transition) | |||
| 358 | + | return | |||
| 359 | + | } | |||
| 360 | + | fresh, admitted := token.ac.beginAttachmentEffect(transition.published) | |||
| 361 | + | if !admitted { | |||
| 362 | + | d.abortPublishedAttachmentTransition(transition) | |||
| 363 | + | return | |||
| 364 | + | } | |||
| 365 | + | freshToken := fresh.connectionToken() | |||
| 366 | + | ac := token.ac | |||
| 367 | + | ac.sendMu.Lock() | |||
| 368 | + | responseErr := d.sendParkedRouteResponseLocked(ac, freshToken.transport, ports.ParkedRouteResponse{RequestID: request.RequestID, Status: ports.ParkedRouteSwitched}) | |||
| 369 | + | if responseErr == nil { | |||
| 370 | + | ac.releaseParkedOutputLocked() | |||
| 371 | + | } | |||
| 372 | + | ac.sendMu.Unlock() | |||
| 373 | + | if responseErr != nil { | |||
| 374 | + | launch := d.reserveAttachmentSendErrorCleanup(freshToken, freshToken.transport.transport) | |||
| 375 | + | fresh.End() | |||
| 376 | + | launch() | |||
| 377 | + | d.deferAttachmentTransitionCleanups(transition) | |||
| 378 | + | return | |||
| 379 | + | } | |||
| 380 | + | fresh.End() | |||
| 381 | + | d.touchMRU(targetSession) | |||
| 382 | + | d.deferAttachmentTransitionCleanups(transition) | |||
| 383 | + | transition.published.rebase = false | |||
| 384 | + | d.firstPaintForTransition(transition.published) | |||
| 385 | + | } | |||
| 386 | + | ||||
| 387 | + | func (d *Daemon) waitForParkedTargetRestore(parent context.Context, name string) error { | |||
| 388 | + | ctx, cancel := context.WithCancel(parent) | |||
| 389 | + | timer := d.clock.NewTimer(ports.HandshakeTimeout) | |||
| 390 | + | stop := make(chan struct{}) | |||
| 391 | + | go func() { | |||
| 392 | + | select { | |||
| 393 | + | case <-timer.C(): | |||
| 394 | + | cancel() | |||
| 395 | + | case <-ctx.Done(): | |||
| 396 | + | case <-stop: | |||
| 397 | + | } | |||
| 398 | + | }() | |||
| 399 | + | err := d.waitForTargetRestore(ctx, name) | |||
| 400 | + | close(stop) | |||
| 401 | + | timer.Stop() | |||
| 402 | + | cancel() | |||
| 403 | + | return err | |||
| 404 | + | } | |||
| 405 | + | ||||
| 406 | + | func (d *Daemon) transitionParkedRouteTarget(token attachmentConnectionToken, target domain.RemoteSessionTarget) (attachmentTransitionResult, *session, error) { | |||
| 407 | + | d.mu.Lock() | |||
| 408 | + | if d.closing { | |||
| 409 | + | d.mu.Unlock() | |||
| 410 | + | return attachmentTransitionResult{}, nil, errAttachmentTransition | |||
| 411 | + | } | |||
| 412 | + | if live := d.findByNameLocked(target.SessionName); live != nil { | |||
| 413 | + | if live.incarnation != target.LifecycleID { | |||
| 414 | + | d.mu.Unlock() | |||
| 415 | + | return attachmentTransitionResult{}, nil, errAttachmentTransition | |||
| 416 | + | } | |||
| 417 | + | tabIndex, ok := remoteTargetTabIndexLocked(live, target) | |||
| 418 | + | if !ok { | |||
| 419 | + | d.mu.Unlock() | |||
| 420 | + | return attachmentTransitionResult{}, nil, errAttachmentTransition | |||
| 421 | + | } | |||
| 422 | + | live.mu.Lock() | |||
| 423 | + | tabID := domain.TabStableID(live.tabs[tabIndex].stableID) | |||
| 424 | + | live.mu.Unlock() | |||
| 425 | + | d.mu.Unlock() | |||
| 426 | + | transition, err := d.transitionAttachment(attachmentTransitionRequest{ | |||
| 427 | + | source: token.sess, target: live, next: token.ac, | |||
| 428 | + | expectedTransport: token.transport, sourceToken: &token, action: "parked-route-switch", | |||
| 429 | + | expectedTargetLifecycle: &attachmentLifecycleFence{ | |||
| 430 | + | name: target.SessionName, incarnation: target.LifecycleID, checkIncarnation: true, | |||
| 431 | + | tabID: tabID, tabIndex: tabIndex, checkTab: true, | |||
| 432 | + | }, | |||
| 433 | + | activateTargetTab: true, targetTabIndex: tabIndex, ready: true, | |||
| 434 | + | }) | |||
| 435 | + | return transition, live, err | |||
| 436 | + | } | |||
| 437 | + | ||||
| 438 | + | inactive, ok := d.inactive[target.SessionName] | |||
| 439 | + | if !ok || !target.Stopped || inactive.incarnation != target.LifecycleID || !inactive.canResume() { | |||
| 440 | + | d.mu.Unlock() | |||
| 441 | + | return attachmentTransitionResult{}, nil, errAttachmentTransition | |||
| 442 | + | } | |||
| 443 | + | tabIndex, ok := remoteTargetTabIndexInactive(inactive, target) | |||
| 444 | + | if !ok { | |||
| 445 | + | d.mu.Unlock() | |||
| 446 | + | return attachmentTransitionResult{}, nil, errAttachmentTransition | |||
| 447 | + | } | |||
| 448 | + | d.mu.Unlock() | |||
| 449 | + | ||||
| 450 | + | var created *session | |||
| 451 | + | transition, err := d.transitionAttachment(attachmentTransitionRequest{ | |||
| 452 | + | source: token.sess, next: token.ac, | |||
| 453 | + | expectedTransport: token.transport, sourceToken: &token, action: "parked-route-restore", | |||
| 454 | + | activateTargetTab: true, targetTabIndex: tabIndex, ready: true, | |||
| 455 | + | createTargetLocked: func() (*session, error) { | |||
| 456 | + | current, ok := d.inactive[target.SessionName] | |||
| 457 | + | if !ok || current.incarnation != target.LifecycleID || !current.canResume() { | |||
| 458 | + | return nil, errAttachmentTransition | |||
| 459 | + | } | |||
| 460 | + | resolved, ok := remoteTargetTabIndexInactive(current, target) | |||
| 461 | + | if !ok || resolved != tabIndex { | |||
| 462 | + | return nil, errAttachmentTransition | |||
| 463 | + | } | |||
| 464 | + | env := copyEnvironment(d.baseEnv) | |||
| 465 | + | cwd := d.dirOrHome(current.cwd) | |||
| 466 | + | var createErr error | |||
| 467 | + | created, createErr = d.resumeRemoteInactiveSessionLocked(target, cwd, token.ac.sizeSnapshot(), env, current) | |||
| 468 | + | return created, createErr | |||
| 469 | + | }, | |||
| 470 | + | }) | |||
| 471 | + | if err != nil && created != nil { | |||
| 472 | + | _ = d.killSession(created, ports.ReasonSessionKilled, true) | |||
| 473 | + | } | |||
| 474 | + | if err != nil { | |||
| 475 | + | return attachmentTransitionResult{}, created, err | |||
| 476 | + | } | |||
| 477 | + | return transition, created, nil | |||
| 478 | + | } | |||
A internal/usecase/daemon/parked_route_test.go +361 -0
| @@ -0,0 +1,361 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "errors" | |
| 5 | + | "io" | |
| 6 | + | "sync" | |
| 7 | + | "testing" | |
| 8 | + | "time" | |
| 9 | + | ||
| 10 | + | "github.com/stretchr/testify/mock" | |
| 11 | + | "github.com/stretchr/testify/require" | |
| 12 | + | ||
| 13 | + | "github.com/bnema/vev/internal/domain" | |
| 14 | + | "github.com/bnema/vev/internal/ports" | |
| 15 | + | portsmocks "github.com/bnema/vev/internal/ports/mocks" | |
| 16 | + | ) | |
| 17 | + | ||
| 18 | + | type parkedRouteFailTransport struct{} | |
| 19 | + | ||
| 20 | + | func (parkedRouteFailTransport) Send(ports.Frame) error { return errors.New("send failed") } | |
| 21 | + | func (parkedRouteFailTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF } | |
| 22 | + | func (parkedRouteFailTransport) Close() error { return nil } | |
| 23 | + | ||
| 24 | + | type parkedRouteExpiryTransport struct { | |
| 25 | + | sends chan ports.Frame | |
| 26 | + | closed chan struct{} | |
| 27 | + | once sync.Once | |
| 28 | + | } | |
| 29 | + | ||
| 30 | + | func (t *parkedRouteExpiryTransport) Send(frame ports.Frame) error { | |
| 31 | + | t.sends <- frame | |
| 32 | + | return nil | |
| 33 | + | } | |
| 34 | + | ||
| 35 | + | func (t *parkedRouteExpiryTransport) Recv() (ports.Frame, error) { | |
| 36 | + | <-t.closed | |
| 37 | + | return ports.Frame{}, io.EOF | |
| 38 | + | } | |
| 39 | + | ||
| 40 | + | func (t *parkedRouteExpiryTransport) Close() error { | |
| 41 | + | t.once.Do(func() { close(t.closed) }) | |
| 42 | + | return nil | |
| 43 | + | } | |
| 44 | + | ||
| 45 | + | type parkedRouteClock struct { | |
| 46 | + | mu sync.Mutex | |
| 47 | + | now time.Time | |
| 48 | + | } | |
| 49 | + | ||
| 50 | + | func (c *parkedRouteClock) Now() time.Time { | |
| 51 | + | c.mu.Lock() | |
| 52 | + | defer c.mu.Unlock() | |
| 53 | + | return c.now | |
| 54 | + | } | |
| 55 | + | ||
| 56 | + | func (c *parkedRouteClock) NewTimer(time.Duration) ports.Timer { return stubTimer{} } | |
| 57 | + | ||
| 58 | + | func (c *parkedRouteClock) setNow(now time.Time) { | |
| 59 | + | c.mu.Lock() | |
| 60 | + | c.now = now | |
| 61 | + | c.mu.Unlock() | |
| 62 | + | } | |
| 63 | + | ||
| 64 | + | func armAndPrepareParkedRoute(t *testing.T, d *Daemon, source *session, ac *attachedClient, sends chan ports.Frame) ports.ParkedRouteLeaseID { | |
| 65 | + | t.Helper() | |
| 66 | + | ac.navigationCapabilities |= ports.NavigationCapabilityHomePicker | |
| 67 | + | token := source.attachmentToken(ac, ac.transport()) | |
| 68 | + | effect, admitted := ac.beginAttachmentEffect(token) | |
| 69 | + | require.True(t, admitted) | |
| 70 | + | token.effect = effect | |
| 71 | + | require.NoError(t, d.sendNavigationActionForAttachment(token, ports.NavigationOpenHomePicker)) | |
| 72 | + | effect.End() | |
| 73 | + | directiveFrame := awaitFrame(t, sends, ports.MsgNavigationAction) | |
| 74 | + | directive, err := ports.UnmarshalNavigationDirective(directiveFrame.Payload) | |
| 75 | + | require.NoError(t, err) | |
| 76 | + | ||
| 77 | + | prepare := ports.ParkedRouteRequest{RequestID: 1, LeaseID: directive.LeaseID, Action: ports.ParkedRoutePrepare} | |
| 78 | + | require.False(t, d.handleAttachmentClientFrame(source.attachmentToken(ac, ac.transport()), ports.Frame{Type: ports.MsgParkedRouteRequest, Payload: ports.MarshalParkedRouteRequest(prepare)})) | |
| 79 | + | readyFrame := awaitFrame(t, sends, ports.MsgParkedRouteResponse) | |
| 80 | + | ready, err := ports.UnmarshalParkedRouteResponse(readyFrame.Payload) | |
| 81 | + | require.NoError(t, err) | |
| 82 | + | require.Equal(t, ports.ParkedRouteResponse{RequestID: 1, Status: ports.ParkedRouteReady}, ready) | |
| 83 | + | return directive.LeaseID | |
| 84 | + | } | |
| 85 | + | ||
| 86 | + | func TestParkedRouteSwitchesExactLiveTargetOnRetainedTransport(t *testing.T) { | |
| 87 | + | d, source, ac, sends, releases := newManualTabSession(t, 1) | |
| 88 | + | defer releaseAll(releases) | |
| 89 | + | ac.navigationCapabilities = ports.NavigationCapabilityHomePicker | |
| 90 | + | ||
| 91 | + | targetLifecycle := domain.SessionLifecycleID{7} | |
| 92 | + | target := &session{ | |
| 93 | + | sessionCore: sessionCore{id: "target", name: "target", incarnation: targetLifecycle, attachments: make(map[*attachedClient]struct{})}, | |
| 94 | + | ctx: source.ctx, cancel: func() {}, | |
| 95 | + | tabs: []*tab{newTab(nil, domain.Size{Cols: 80, Rows: 23})}, | |
| 96 | + | } | |
| 97 | + | publishTiledPaneOwners(target, target.tabs[0]) | |
| 98 | + | d.mu.Lock() | |
| 99 | + | d.sessions[target.id] = target | |
| 100 | + | d.mu.Unlock() | |
| 101 | + | ac.setRouteSnapshot(ports.RecentRouteSnapshot{Generation: 1}) | |
| 102 | + | ||
| 103 | + | leaseID := armAndPrepareParkedRoute(t, d, source, ac, sends) | |
| 104 | + | require.True(t, ac.parkedRouteOutput.Load()) | |
| 105 | + | ||
| 106 | + | targetRequest := domain.RemoteSessionTarget{ | |
| 107 | + | Endpoint: "remote", DisplayOrigin: "remote", LifecycleID: targetLifecycle, | |
| 108 | + | SessionName: "target", LiveTabID: domain.TabStableID(target.tabs[0].stableID), | |
| 109 | + | } | |
| 110 | + | switchRequest := ports.ParkedRouteRequest{RequestID: 2, LeaseID: leaseID, Action: ports.ParkedRouteSwitch, Target: &targetRequest} | |
| 111 | + | require.False(t, d.handleAttachmentClientFrame(source.attachmentToken(ac, ac.transport()), ports.Frame{Type: ports.MsgParkedRouteRequest, Payload: ports.MarshalParkedRouteRequest(switchRequest)})) | |
| 112 | + | ||
| 113 | + | require.Same(t, target, ac.currentAttachmentSession()) | |
| 114 | + | require.False(t, ac.parkedRouteOutput.Load()) | |
| 115 | + | responseFrame := awaitFrame(t, sends, ports.MsgParkedRouteResponse) | |
| 116 | + | response, err := ports.UnmarshalParkedRouteResponse(responseFrame.Payload) | |
| 117 | + | require.NoError(t, err) | |
| 118 | + | require.Equal(t, ports.ParkedRouteResponse{RequestID: 2, Status: ports.ParkedRouteSwitched}, response) | |
| 119 | + | } | |
| 120 | + | ||
| 121 | + | func TestParkedRouteSwitchAcceptedBeforeLeaseExpiryCommitsAtomically(t *testing.T) { | |
| 122 | + | d, source, ac, sends, releases := newManualTabSession(t, 1) | |
| 123 | + | defer releaseAll(releases) | |
| 124 | + | clock := &parkedRouteClock{} | |
| 125 | + | d.clock = clock | |
| 126 | + | ac.navigationCapabilities = ports.NavigationCapabilityHomePicker | |
| 127 | + | ||
| 128 | + | targetLifecycle := domain.SessionLifecycleID{7} | |
| 129 | + | target := &session{ | |
| 130 | + | sessionCore: sessionCore{id: "target", name: "target", incarnation: targetLifecycle, attachments: make(map[*attachedClient]struct{})}, | |
| 131 | + | ctx: source.ctx, cancel: func() {}, | |
| 132 | + | tabs: []*tab{newTab(nil, domain.Size{Cols: 80, Rows: 23})}, | |
| 133 | + | } | |
| 134 | + | publishTiledPaneOwners(target, target.tabs[0]) | |
| 135 | + | d.mu.Lock() | |
| 136 | + | d.sessions[target.id] = target | |
| 137 | + | d.mu.Unlock() | |
| 138 | + | ac.setRouteSnapshot(ports.RecentRouteSnapshot{Generation: 1}) | |
| 139 | + | ||
| 140 | + | leaseID := armAndPrepareParkedRoute(t, d, source, ac, sends) | |
| 141 | + | d.afterAttachmentEffectsFrozen = func() { | |
| 142 | + | clock.setNow(time.Time{}.Add(parkedRouteLeaseTTL)) | |
| 143 | + | } | |
| 144 | + | ||
| 145 | + | targetRequest := domain.RemoteSessionTarget{ | |
| 146 | + | Endpoint: "remote", DisplayOrigin: "remote", LifecycleID: targetLifecycle, | |
| 147 | + | SessionName: "target", LiveTabID: domain.TabStableID(target.tabs[0].stableID), | |
| 148 | + | } | |
| 149 | + | switchRequest := ports.ParkedRouteRequest{RequestID: 2, LeaseID: leaseID, Action: ports.ParkedRouteSwitch, Target: &targetRequest} | |
| 150 | + | d.handleAttachmentClientFrame(source.attachmentToken(ac, ac.transport()), ports.Frame{Type: ports.MsgParkedRouteRequest, Payload: ports.MarshalParkedRouteRequest(switchRequest)}) | |
| 151 | + | ||
| 152 | + | require.Same(t, target, ac.currentAttachmentSession()) | |
| 153 | + | responseFrame := awaitFrame(t, sends, ports.MsgParkedRouteResponse) | |
| 154 | + | response, err := ports.UnmarshalParkedRouteResponse(responseFrame.Payload) | |
| 155 | + | require.NoError(t, err) | |
| 156 | + | require.Equal(t, ports.ParkedRouteSwitched, response.Status) | |
| 157 | + | } | |
| 158 | + | ||
| 159 | + | func TestParkedRouteRestoresExactStoppedTargetWithDaemonEnvironment(t *testing.T) { | |
| 160 | + | d := newTestDaemon(t, newFactory(t, newQuietPTY()), stubClock{}) | |
| 161 | + | d.baseEnv = []string{"OWNER=daemon"} | |
| 162 | + | source, ac, sends := addRemoteRefreshPickerOwner(t, d, "source") | |
| 163 | + | ac.navigationCapabilities = ports.NavigationCapabilityHomePicker | |
| 164 | + | source.mu.Lock() | |
| 165 | + | source.env = []string{"OWNER=source"} | |
| 166 | + | source.mu.Unlock() | |
| 167 | + | ||
| 168 | + | lifecycle := domain.SessionLifecycleID{8} | |
| 169 | + | d.mu.Lock() | |
| 170 | + | d.inactive["stopped"] = inactiveSession{ | |
| 171 | + | name: "stopped", cwd: "/remote/work", incarnation: lifecycle, state: ports.SessionDown, | |
| 172 | + | tabNames: []string{"alpha", "beta"}, | |
| 173 | + | tabRecords: []domain.CatalogueTabRecord{ | |
| 174 | + | {StableID: "tab-a", Name: "alpha"}, | |
| 175 | + | {StableID: "tab-b", Name: "beta"}, | |
| 176 | + | }, | |
| 177 | + | } | |
| 178 | + | d.mu.Unlock() | |
| 179 | + | ac.setRouteSnapshot(ports.RecentRouteSnapshot{Generation: 1}) | |
| 180 | + | ||
| 181 | + | leaseID := armAndPrepareParkedRoute(t, d, source, ac, sends) | |
| 182 | + | ||
| 183 | + | target := domain.RemoteSessionTarget{ | |
| 184 | + | Endpoint: "remote", DisplayOrigin: "remote", LifecycleID: lifecycle, | |
| 185 | + | SessionName: "stopped", Stopped: true, StoppedTab: domain.NewStableTabSelector("tab-b"), | |
| 186 | + | } | |
| 187 | + | switchRequest := ports.ParkedRouteRequest{RequestID: 2, LeaseID: leaseID, Action: ports.ParkedRouteSwitch, Target: &target} | |
| 188 | + | d.handleAttachmentClientFrame(source.attachmentToken(ac, ac.transport()), ports.Frame{Type: ports.MsgParkedRouteRequest, Payload: ports.MarshalParkedRouteRequest(switchRequest)}) | |
| 189 | + | ||
| 190 | + | restored := ac.currentAttachmentSession() | |
| 191 | + | require.NotNil(t, restored) | |
| 192 | + | require.Equal(t, "stopped", restored.name) | |
| 193 | + | require.Equal(t, lifecycle, restored.incarnation) | |
| 194 | + | require.Equal(t, domain.TabStableID("tab-b"), ac.viewSnapshot().tabID) | |
| 195 | + | restored.mu.Lock() | |
| 196 | + | require.Equal(t, []string{"OWNER=daemon"}, restored.env) | |
| 197 | + | restored.mu.Unlock() | |
| 198 | + | } | |
| 199 | + | ||
| 200 | + | func TestParkedRouteStaleTargetLeavesSourceResumable(t *testing.T) { | |
| 201 | + | d, source, ac, sends, releases := newManualTabSession(t, 1) | |
| 202 | + | defer releaseAll(releases) | |
| 203 | + | ac.navigationCapabilities = ports.NavigationCapabilityHomePicker | |
| 204 | + | ac.setRouteSnapshot(ports.RecentRouteSnapshot{Generation: 1}) | |
| 205 | + | ||
| 206 | + | leaseID := armAndPrepareParkedRoute(t, d, source, ac, sends) | |
| 207 | + | ||
| 208 | + | missing := domain.RemoteSessionTarget{ | |
| 209 | + | Endpoint: "remote", DisplayOrigin: "remote", LifecycleID: domain.SessionLifecycleID{9}, | |
| 210 | + | SessionName: "missing", LiveTabID: "tab-1", | |
| 211 | + | } | |
| 212 | + | switchRequest := ports.ParkedRouteRequest{RequestID: 2, LeaseID: leaseID, Action: ports.ParkedRouteSwitch, Target: &missing} | |
| 213 | + | d.handleAttachmentClientFrame(source.attachmentToken(ac, ac.transport()), ports.Frame{Type: ports.MsgParkedRouteRequest, Payload: ports.MarshalParkedRouteRequest(switchRequest)}) | |
| 214 | + | staleFrame := awaitFrame(t, sends, ports.MsgParkedRouteResponse) | |
| 215 | + | stale, err := ports.UnmarshalParkedRouteResponse(staleFrame.Payload) | |
| 216 | + | require.NoError(t, err) | |
| 217 | + | require.Equal(t, ports.ParkedRouteStaleTarget, stale.Status) | |
| 218 | + | require.Same(t, source, ac.currentAttachmentSession()) | |
| 219 | + | require.True(t, ac.parkedRouteOutput.Load()) | |
| 220 | + | ||
| 221 | + | resume := ports.ParkedRouteRequest{RequestID: 3, LeaseID: leaseID, Action: ports.ParkedRouteResume} | |
| 222 | + | d.handleAttachmentClientFrame(source.attachmentToken(ac, ac.transport()), ports.Frame{Type: ports.MsgParkedRouteRequest, Payload: ports.MarshalParkedRouteRequest(resume)}) | |
| 223 | + | resumedFrame := awaitFrame(t, sends, ports.MsgParkedRouteResponse) | |
| 224 | + | resumed, err := ports.UnmarshalParkedRouteResponse(resumedFrame.Payload) | |
| 225 | + | require.NoError(t, err) | |
| 226 | + | require.Equal(t, ports.ParkedRouteResumed, resumed.Status) | |
| 227 | + | require.False(t, ac.parkedRouteOutput.Load()) | |
| 228 | + | outputFrame := awaitFrame(t, sends, ports.MsgOutput) | |
| 229 | + | output, err := ports.UnmarshalOutput(outputFrame.Payload) | |
| 230 | + | require.NoError(t, err) | |
| 231 | + | require.True(t, output.Full) | |
| 232 | + | } | |
| 233 | + | ||
| 234 | + | func TestParkedRouteDoesNotReplayOneShotEffectsAfterResume(t *testing.T) { | |
| 235 | + | d, source, ac, sends, releases := newManualTabSession(t, 1) | |
| 236 | + | defer releaseAll(releases) | |
| 237 | + | ac.navigationCapabilities = ports.NavigationCapabilityHomePicker | |
| 238 | + | ||
| 239 | + | leaseID := armAndPrepareParkedRoute(t, d, source, ac, sends) | |
| 240 | + | require.NoError(t, d.boundedSendOutputErr(ac, []byte("one-shot effect"))) | |
| 241 | + | select { | |
| 242 | + | case frame := <-sends: | |
| 243 | + | t.Fatalf("parked one-shot effect emitted frame type %d", frame.Type) | |
| 244 | + | default: | |
| 245 | + | } | |
| 246 | + | ||
| 247 | + | resume := ports.ParkedRouteRequest{RequestID: 2, LeaseID: leaseID, Action: ports.ParkedRouteResume} | |
| 248 | + | d.handleAttachmentClientFrame(source.attachmentToken(ac, ac.transport()), ports.Frame{Type: ports.MsgParkedRouteRequest, Payload: ports.MarshalParkedRouteRequest(resume)}) | |
| 249 | + | _ = awaitFrame(t, sends, ports.MsgParkedRouteResponse) | |
| 250 | + | outputFrame := awaitFrame(t, sends, ports.MsgOutput) | |
| 251 | + | output, err := ports.UnmarshalOutput(outputFrame.Payload) | |
| 252 | + | require.NoError(t, err) | |
| 253 | + | require.True(t, output.Full) | |
| 254 | + | require.NotContains(t, string(output.Data), "one-shot effect") | |
| 255 | + | select { | |
| 256 | + | case frame := <-sends: | |
| 257 | + | t.Fatalf("parked one-shot effect replayed as frame type %d", frame.Type) | |
| 258 | + | default: | |
| 259 | + | } | |
| 260 | + | } | |
| 261 | + | ||
| 262 | + | func TestParkedRouteExpiredLeaseFailsClosed(t *testing.T) { | |
| 263 | + | d, source, ac, sends, releases := newManualTabSession(t, 1) | |
| 264 | + | defer releaseAll(releases) | |
| 265 | + | ac.navigationCapabilities = ports.NavigationCapabilityHomePicker | |
| 266 | + | ||
| 267 | + | leaseID := armAndPrepareParkedRoute(t, d, source, ac, sends) | |
| 268 | + | ||
| 269 | + | ac.parkedRouteMu.Lock() | |
| 270 | + | ac.parkedRoute.expiresAt = d.clock.Now() | |
| 271 | + | ac.parkedRouteMu.Unlock() | |
| 272 | + | resume := ports.ParkedRouteRequest{RequestID: 2, LeaseID: leaseID, Action: ports.ParkedRouteResume} | |
| 273 | + | d.handleAttachmentClientFrame(source.attachmentToken(ac, ac.transport()), ports.Frame{Type: ports.MsgParkedRouteRequest, Payload: ports.MarshalParkedRouteRequest(resume)}) | |
| 274 | + | responseFrame := awaitFrame(t, sends, ports.MsgParkedRouteResponse) | |
| 275 | + | response, err := ports.UnmarshalParkedRouteResponse(responseFrame.Payload) | |
| 276 | + | require.NoError(t, err) | |
| 277 | + | require.Equal(t, ports.ParkedRouteExpired, response.Status) | |
| 278 | + | require.Same(t, source, ac.currentAttachmentSession()) | |
| 279 | + | require.False(t, ac.parkedRouteOutput.Load()) | |
| 280 | + | } | |
| 281 | + | ||
| 282 | + | func TestParkedRouteLeaseExpiryClosesExactRetainedTransport(t *testing.T) { | |
| 283 | + | d, source, ac, _, releases := newManualTabSession(t, 1) | |
| 284 | + | defer releaseAll(releases) | |
| 285 | + | expiry := make(chan time.Time, 1) | |
| 286 | + | timer := portsmocks.NewMockTimer(t) | |
| 287 | + | timer.EXPECT().C().Return(expiry).Maybe() | |
| 288 | + | timer.EXPECT().Stop().Return(true).Maybe() | |
| 289 | + | clock := portsmocks.NewMockClock(t) | |
| 290 | + | clock.EXPECT().Now().Return(time.Time{}).Maybe() | |
| 291 | + | clock.EXPECT().NewTimer(mock.Anything).Return(timer).Once() | |
| 292 | + | d.clock = clock | |
| 293 | + | transport := &parkedRouteExpiryTransport{sends: make(chan ports.Frame, 8), closed: make(chan struct{})} | |
| 294 | + | ac.replaceTransport(transport) | |
| 295 | + | ||
| 296 | + | _ = armAndPrepareParkedRoute(t, d, source, ac, transport.sends) | |
| 297 | + | expiry <- time.Time{}.Add(parkedRouteLeaseTTL) | |
| 298 | + | ||
| 299 | + | select { | |
| 300 | + | case <-transport.closed: | |
| 301 | + | case <-time.After(time.Second): | |
| 302 | + | t.Fatal("parked transport remained open after its lease expired") | |
| 303 | + | } | |
| 304 | + | } | |
| 305 | + | ||
| 306 | + | func TestParkedRouteMalformedSwitchRequestsFailClosed(t *testing.T) { | |
| 307 | + | d, source, ac, sends, releases := newManualTabSession(t, 1) | |
| 308 | + | defer releaseAll(releases) | |
| 309 | + | leaseID := ports.ParkedRouteLeaseID{1} | |
| 310 | + | for _, request := range []ports.ParkedRouteRequest{ | |
| 311 | + | {RequestID: 1, LeaseID: leaseID, Action: ports.ParkedRouteSwitch}, | |
| 312 | + | {RequestID: 2, Action: ports.ParkedRouteSwitch, Target: &domain.RemoteSessionTarget{}}, | |
| 313 | + | {RequestID: 3, LeaseID: leaseID, Action: 99}, | |
| 314 | + | } { | |
| 315 | + | d.handleParkedRouteRequest(source.attachmentToken(ac, ac.transport()), request) | |
| 316 | + | } | |
| 317 | + | require.Same(t, source, ac.currentAttachmentSession()) | |
| 318 | + | select { | |
| 319 | + | case frame := <-sends: | |
| 320 | + | t.Fatalf("malformed parked route emitted frame type %d", frame.Type) | |
| 321 | + | default: | |
| 322 | + | } | |
| 323 | + | } | |
| 324 | + | ||
| 325 | + | func TestParkedRouteResponseFailureDoesNotDrainItsOwnFrameEffect(t *testing.T) { | |
| 326 | + | d, source, ac, _, releases := newManualTabSession(t, 1) | |
| 327 | + | defer releaseAll(releases) | |
| 328 | + | ac.replaceTransport(parkedRouteFailTransport{}) | |
| 329 | + | token := source.attachmentToken(ac, ac.transport()) | |
| 330 | + | effect, admitted := ac.beginAttachmentEffect(token) | |
| 331 | + | require.True(t, admitted) | |
| 332 | + | token.effect = effect | |
| 333 | + | ||
| 334 | + | result := make(chan bool, 1) | |
| 335 | + | go func() { | |
| 336 | + | result <- d.respondParkedRoute(token, ports.ParkedRouteResponse{RequestID: 1, Status: ports.ParkedRouteReady}) | |
| 337 | + | }() | |
| 338 | + | require.False(t, awaitTestValue(t, result, "parked-route send failure cleanup")) | |
| 339 | + | d.attachmentCleanupWg.Wait() | |
| 340 | + | } | |
| 341 | + | ||
| 342 | + | func TestParkedRouteDetachClearsLeaseAndSuspension(t *testing.T) { | |
| 343 | + | d, source, ac, sends, releases := newManualTabSession(t, 1) | |
| 344 | + | defer releaseAll(releases) | |
| 345 | + | ac.navigationCapabilities = ports.NavigationCapabilityHomePicker | |
| 346 | + | ||
| 347 | + | _ = armAndPrepareParkedRoute(t, d, source, ac, sends) | |
| 348 | + | require.True(t, ac.parkedRouteOutput.Load()) | |
| 349 | + | require.NoError(t, d.boundedSendOutputErr(ac, []byte("suppressed side effect"))) | |
| 350 | + | select { | |
| 351 | + | case frame := <-sends: | |
| 352 | + | t.Fatalf("parked side effect emitted frame type %d", frame.Type) | |
| 353 | + | default: | |
| 354 | + | } | |
| 355 | + | ||
| 356 | + | d.clientGone(source, ac, ac.transport(), true) | |
| 357 | + | ac.parkedRouteMu.Lock() | |
| 358 | + | require.Nil(t, ac.parkedRoute) | |
| 359 | + | ac.parkedRouteMu.Unlock() | |
| 360 | + | require.False(t, ac.parkedRouteOutput.Load()) | |
| 361 | + | } | |
| @@ -0,0 +1,361 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "errors" | |||
| 5 | + | "io" | |||
| 6 | + | "sync" | |||
| 7 | + | "testing" | |||
| 8 | + | "time" | |||
| 9 | + | ||||
| 10 | + | "github.com/stretchr/testify/mock" | |||
| 11 | + | "github.com/stretchr/testify/require" | |||
| 12 | + | ||||
| 13 | + | "github.com/bnema/vev/internal/domain" | |||
| 14 | + | "github.com/bnema/vev/internal/ports" | |||
| 15 | + | portsmocks "github.com/bnema/vev/internal/ports/mocks" | |||
| 16 | + | ) | |||
| 17 | + | ||||
| 18 | + | type parkedRouteFailTransport struct{} | |||
| 19 | + | ||||
| 20 | + | func (parkedRouteFailTransport) Send(ports.Frame) error { return errors.New("send failed") } | |||
| 21 | + | func (parkedRouteFailTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF } | |||
| 22 | + | func (parkedRouteFailTransport) Close() error { return nil } | |||
| 23 | + | ||||
| 24 | + | type parkedRouteExpiryTransport struct { | |||
| 25 | + | sends chan ports.Frame | |||
| 26 | + | closed chan struct{} | |||
| 27 | + | once sync.Once | |||
| 28 | + | } | |||
| 29 | + | ||||
| 30 | + | func (t *parkedRouteExpiryTransport) Send(frame ports.Frame) error { | |||
| 31 | + | t.sends <- frame | |||
| 32 | + | return nil | |||
| 33 | + | } | |||
| 34 | + | ||||
| 35 | + | func (t *parkedRouteExpiryTransport) Recv() (ports.Frame, error) { | |||
| 36 | + | <-t.closed | |||
| 37 | + | return ports.Frame{}, io.EOF | |||
| 38 | + | } | |||
| 39 | + | ||||
| 40 | + | func (t *parkedRouteExpiryTransport) Close() error { | |||
| 41 | + | t.once.Do(func() { close(t.closed) }) | |||
| 42 | + | return nil | |||
| 43 | + | } | |||
| 44 | + | ||||
| 45 | + | type parkedRouteClock struct { | |||
| 46 | + | mu sync.Mutex | |||
| 47 | + | now time.Time | |||
| 48 | + | } | |||
| 49 | + | ||||
| 50 | + | func (c *parkedRouteClock) Now() time.Time { | |||
| 51 | + | c.mu.Lock() | |||
| 52 | + | defer c.mu.Unlock() | |||
| 53 | + | return c.now | |||
| 54 | + | } | |||
| 55 | + | ||||
| 56 | + | func (c *parkedRouteClock) NewTimer(time.Duration) ports.Timer { return stubTimer{} } | |||
| 57 | + | ||||
| 58 | + | func (c *parkedRouteClock) setNow(now time.Time) { | |||
| 59 | + | c.mu.Lock() | |||
| 60 | + | c.now = now | |||
| 61 | + | c.mu.Unlock() | |||
| 62 | + | } | |||
| 63 | + | ||||
| 64 | + | func armAndPrepareParkedRoute(t *testing.T, d *Daemon, source *session, ac *attachedClient, sends chan ports.Frame) ports.ParkedRouteLeaseID { | |||
| 65 | + | t.Helper() | |||
| 66 | + | ac.navigationCapabilities |= ports.NavigationCapabilityHomePicker | |||
| 67 | + | token := source.attachmentToken(ac, ac.transport()) | |||
| 68 | + | effect, admitted := ac.beginAttachmentEffect(token) | |||
| 69 | + | require.True(t, admitted) | |||
| 70 | + | token.effect = effect | |||
| 71 | + | require.NoError(t, d.sendNavigationActionForAttachment(token, ports.NavigationOpenHomePicker)) | |||
| 72 | + | effect.End() | |||
| 73 | + | directiveFrame := awaitFrame(t, sends, ports.MsgNavigationAction) | |||
| 74 | + | directive, err := ports.UnmarshalNavigationDirective(directiveFrame.Payload) | |||
| 75 | + | require.NoError(t, err) | |||
| 76 | + | ||||
| 77 | + | prepare := ports.ParkedRouteRequest{RequestID: 1, LeaseID: directive.LeaseID, Action: ports.ParkedRoutePrepare} | |||
| 78 | + | require.False(t, d.handleAttachmentClientFrame(source.attachmentToken(ac, ac.transport()), ports.Frame{Type: ports.MsgParkedRouteRequest, Payload: ports.MarshalParkedRouteRequest(prepare)})) | |||
| 79 | + | readyFrame := awaitFrame(t, sends, ports.MsgParkedRouteResponse) | |||
| 80 | + | ready, err := ports.UnmarshalParkedRouteResponse(readyFrame.Payload) | |||
| 81 | + | require.NoError(t, err) | |||
| 82 | + | require.Equal(t, ports.ParkedRouteResponse{RequestID: 1, Status: ports.ParkedRouteReady}, ready) | |||
| 83 | + | return directive.LeaseID | |||
| 84 | + | } | |||
| 85 | + | ||||
| 86 | + | func TestParkedRouteSwitchesExactLiveTargetOnRetainedTransport(t *testing.T) { | |||
| 87 | + | d, source, ac, sends, releases := newManualTabSession(t, 1) | |||
| 88 | + | defer releaseAll(releases) | |||
| 89 | + | ac.navigationCapabilities = ports.NavigationCapabilityHomePicker | |||
| 90 | + | ||||
| 91 | + | targetLifecycle := domain.SessionLifecycleID{7} | |||
| 92 | + | target := &session{ | |||
| 93 | + | sessionCore: sessionCore{id: "target", name: "target", incarnation: targetLifecycle, attachments: make(map[*attachedClient]struct{})}, | |||
| 94 | + | ctx: source.ctx, cancel: func() {}, | |||
| 95 | + | tabs: []*tab{newTab(nil, domain.Size{Cols: 80, Rows: 23})}, | |||
| 96 | + | } | |||
| 97 | + | publishTiledPaneOwners(target, target.tabs[0]) | |||
| 98 | + | d.mu.Lock() | |||
| 99 | + | d.sessions[target.id] = target | |||
| 100 | + | d.mu.Unlock() | |||
| 101 | + | ac.setRouteSnapshot(ports.RecentRouteSnapshot{Generation: 1}) | |||
| 102 | + | ||||
| 103 | + | leaseID := armAndPrepareParkedRoute(t, d, source, ac, sends) | |||
| 104 | + | require.True(t, ac.parkedRouteOutput.Load()) | |||
| 105 | + | ||||
| 106 | + | targetRequest := domain.RemoteSessionTarget{ | |||
| 107 | + | Endpoint: "remote", DisplayOrigin: "remote", LifecycleID: targetLifecycle, | |||
| 108 | + | SessionName: "target", LiveTabID: domain.TabStableID(target.tabs[0].stableID), | |||
| 109 | + | } | |||
| 110 | + | switchRequest := ports.ParkedRouteRequest{RequestID: 2, LeaseID: leaseID, Action: ports.ParkedRouteSwitch, Target: &targetRequest} | |||
| 111 | + | require.False(t, d.handleAttachmentClientFrame(source.attachmentToken(ac, ac.transport()), ports.Frame{Type: ports.MsgParkedRouteRequest, Payload: ports.MarshalParkedRouteRequest(switchRequest)})) | |||
| 112 | + | ||||
| 113 | + | require.Same(t, target, ac.currentAttachmentSession()) | |||
| 114 | + | require.False(t, ac.parkedRouteOutput.Load()) | |||
| 115 | + | responseFrame := awaitFrame(t, sends, ports.MsgParkedRouteResponse) | |||
| 116 | + | response, err := ports.UnmarshalParkedRouteResponse(responseFrame.Payload) | |||
| 117 | + | require.NoError(t, err) | |||
| 118 | + | require.Equal(t, ports.ParkedRouteResponse{RequestID: 2, Status: ports.ParkedRouteSwitched}, response) | |||
| 119 | + | } | |||
| 120 | + | ||||
| 121 | + | func TestParkedRouteSwitchAcceptedBeforeLeaseExpiryCommitsAtomically(t *testing.T) { | |||
| 122 | + | d, source, ac, sends, releases := newManualTabSession(t, 1) | |||
| 123 | + | defer releaseAll(releases) | |||
| 124 | + | clock := &parkedRouteClock{} | |||
| 125 | + | d.clock = clock | |||
| 126 | + | ac.navigationCapabilities = ports.NavigationCapabilityHomePicker | |||
| 127 | + | ||||
| 128 | + | targetLifecycle := domain.SessionLifecycleID{7} | |||
| 129 | + | target := &session{ | |||
| 130 | + | sessionCore: sessionCore{id: "target", name: "target", incarnation: targetLifecycle, attachments: make(map[*attachedClient]struct{})}, | |||
| 131 | + | ctx: source.ctx, cancel: func() {}, | |||
| 132 | + | tabs: []*tab{newTab(nil, domain.Size{Cols: 80, Rows: 23})}, | |||
| 133 | + | } | |||
| 134 | + | publishTiledPaneOwners(target, target.tabs[0]) | |||
| 135 | + | d.mu.Lock() | |||
| 136 | + | d.sessions[target.id] = target | |||
| 137 | + | d.mu.Unlock() | |||
| 138 | + | ac.setRouteSnapshot(ports.RecentRouteSnapshot{Generation: 1}) | |||
| 139 | + | ||||
| 140 | + | leaseID := armAndPrepareParkedRoute(t, d, source, ac, sends) | |||
| 141 | + | d.afterAttachmentEffectsFrozen = func() { | |||
| 142 | + | clock.setNow(time.Time{}.Add(parkedRouteLeaseTTL)) | |||
| 143 | + | } | |||
| 144 | + | ||||
| 145 | + | targetRequest := domain.RemoteSessionTarget{ | |||
| 146 | + | Endpoint: "remote", DisplayOrigin: "remote", LifecycleID: targetLifecycle, | |||
| 147 | + | SessionName: "target", LiveTabID: domain.TabStableID(target.tabs[0].stableID), | |||
| 148 | + | } | |||
| 149 | + | switchRequest := ports.ParkedRouteRequest{RequestID: 2, LeaseID: leaseID, Action: ports.ParkedRouteSwitch, Target: &targetRequest} | |||
| 150 | + | d.handleAttachmentClientFrame(source.attachmentToken(ac, ac.transport()), ports.Frame{Type: ports.MsgParkedRouteRequest, Payload: ports.MarshalParkedRouteRequest(switchRequest)}) | |||
| 151 | + | ||||
| 152 | + | require.Same(t, target, ac.currentAttachmentSession()) | |||
| 153 | + | responseFrame := awaitFrame(t, sends, ports.MsgParkedRouteResponse) | |||
| 154 | + | response, err := ports.UnmarshalParkedRouteResponse(responseFrame.Payload) | |||
| 155 | + | require.NoError(t, err) | |||
| 156 | + | require.Equal(t, ports.ParkedRouteSwitched, response.Status) | |||
| 157 | + | } | |||
| 158 | + | ||||
| 159 | + | func TestParkedRouteRestoresExactStoppedTargetWithDaemonEnvironment(t *testing.T) { | |||
| 160 | + | d := newTestDaemon(t, newFactory(t, newQuietPTY()), stubClock{}) | |||
| 161 | + | d.baseEnv = []string{"OWNER=daemon"} | |||
| 162 | + | source, ac, sends := addRemoteRefreshPickerOwner(t, d, "source") | |||
| 163 | + | ac.navigationCapabilities = ports.NavigationCapabilityHomePicker | |||
| 164 | + | source.mu.Lock() | |||
| 165 | + | source.env = []string{"OWNER=source"} | |||
| 166 | + | source.mu.Unlock() | |||
| 167 | + | ||||
| 168 | + | lifecycle := domain.SessionLifecycleID{8} | |||
| 169 | + | d.mu.Lock() | |||
| 170 | + | d.inactive["stopped"] = inactiveSession{ | |||
| 171 | + | name: "stopped", cwd: "/remote/work", incarnation: lifecycle, state: ports.SessionDown, | |||
| 172 | + | tabNames: []string{"alpha", "beta"}, | |||
| 173 | + | tabRecords: []domain.CatalogueTabRecord{ | |||
| 174 | + | {StableID: "tab-a", Name: "alpha"}, | |||
| 175 | + | {StableID: "tab-b", Name: "beta"}, | |||
| 176 | + | }, | |||
| 177 | + | } | |||
| 178 | + | d.mu.Unlock() | |||
| 179 | + | ac.setRouteSnapshot(ports.RecentRouteSnapshot{Generation: 1}) | |||
| 180 | + | ||||
| 181 | + | leaseID := armAndPrepareParkedRoute(t, d, source, ac, sends) | |||
| 182 | + | ||||
| 183 | + | target := domain.RemoteSessionTarget{ | |||
| 184 | + | Endpoint: "remote", DisplayOrigin: "remote", LifecycleID: lifecycle, | |||
| 185 | + | SessionName: "stopped", Stopped: true, StoppedTab: domain.NewStableTabSelector("tab-b"), | |||
| 186 | + | } | |||
| 187 | + | switchRequest := ports.ParkedRouteRequest{RequestID: 2, LeaseID: leaseID, Action: ports.ParkedRouteSwitch, Target: &target} | |||
| 188 | + | d.handleAttachmentClientFrame(source.attachmentToken(ac, ac.transport()), ports.Frame{Type: ports.MsgParkedRouteRequest, Payload: ports.MarshalParkedRouteRequest(switchRequest)}) | |||
| 189 | + | ||||
| 190 | + | restored := ac.currentAttachmentSession() | |||
| 191 | + | require.NotNil(t, restored) | |||
| 192 | + | require.Equal(t, "stopped", restored.name) | |||
| 193 | + | require.Equal(t, lifecycle, restored.incarnation) | |||
| 194 | + | require.Equal(t, domain.TabStableID("tab-b"), ac.viewSnapshot().tabID) | |||
| 195 | + | restored.mu.Lock() | |||
| 196 | + | require.Equal(t, []string{"OWNER=daemon"}, restored.env) | |||
| 197 | + | restored.mu.Unlock() | |||
| 198 | + | } | |||
| 199 | + | ||||
| 200 | + | func TestParkedRouteStaleTargetLeavesSourceResumable(t *testing.T) { | |||
| 201 | + | d, source, ac, sends, releases := newManualTabSession(t, 1) | |||
| 202 | + | defer releaseAll(releases) | |||
| 203 | + | ac.navigationCapabilities = ports.NavigationCapabilityHomePicker | |||
| 204 | + | ac.setRouteSnapshot(ports.RecentRouteSnapshot{Generation: 1}) | |||
| 205 | + | ||||
| 206 | + | leaseID := armAndPrepareParkedRoute(t, d, source, ac, sends) | |||
| 207 | + | ||||
| 208 | + | missing := domain.RemoteSessionTarget{ | |||
| 209 | + | Endpoint: "remote", DisplayOrigin: "remote", LifecycleID: domain.SessionLifecycleID{9}, | |||
| 210 | + | SessionName: "missing", LiveTabID: "tab-1", | |||
| 211 | + | } | |||
| 212 | + | switchRequest := ports.ParkedRouteRequest{RequestID: 2, LeaseID: leaseID, Action: ports.ParkedRouteSwitch, Target: &missing} | |||
| 213 | + | d.handleAttachmentClientFrame(source.attachmentToken(ac, ac.transport()), ports.Frame{Type: ports.MsgParkedRouteRequest, Payload: ports.MarshalParkedRouteRequest(switchRequest)}) | |||
| 214 | + | staleFrame := awaitFrame(t, sends, ports.MsgParkedRouteResponse) | |||
| 215 | + | stale, err := ports.UnmarshalParkedRouteResponse(staleFrame.Payload) | |||
| 216 | + | require.NoError(t, err) | |||
| 217 | + | require.Equal(t, ports.ParkedRouteStaleTarget, stale.Status) | |||
| 218 | + | require.Same(t, source, ac.currentAttachmentSession()) | |||
| 219 | + | require.True(t, ac.parkedRouteOutput.Load()) | |||
| 220 | + | ||||
| 221 | + | resume := ports.ParkedRouteRequest{RequestID: 3, LeaseID: leaseID, Action: ports.ParkedRouteResume} | |||
| 222 | + | d.handleAttachmentClientFrame(source.attachmentToken(ac, ac.transport()), ports.Frame{Type: ports.MsgParkedRouteRequest, Payload: ports.MarshalParkedRouteRequest(resume)}) | |||
| 223 | + | resumedFrame := awaitFrame(t, sends, ports.MsgParkedRouteResponse) | |||
| 224 | + | resumed, err := ports.UnmarshalParkedRouteResponse(resumedFrame.Payload) | |||
| 225 | + | require.NoError(t, err) | |||
| 226 | + | require.Equal(t, ports.ParkedRouteResumed, resumed.Status) | |||
| 227 | + | require.False(t, ac.parkedRouteOutput.Load()) | |||
| 228 | + | outputFrame := awaitFrame(t, sends, ports.MsgOutput) | |||
| 229 | + | output, err := ports.UnmarshalOutput(outputFrame.Payload) | |||
| 230 | + | require.NoError(t, err) | |||
| 231 | + | require.True(t, output.Full) | |||
| 232 | + | } | |||
| 233 | + | ||||
| 234 | + | func TestParkedRouteDoesNotReplayOneShotEffectsAfterResume(t *testing.T) { | |||
| 235 | + | d, source, ac, sends, releases := newManualTabSession(t, 1) | |||
| 236 | + | defer releaseAll(releases) | |||
| 237 | + | ac.navigationCapabilities = ports.NavigationCapabilityHomePicker | |||
| 238 | + | ||||
| 239 | + | leaseID := armAndPrepareParkedRoute(t, d, source, ac, sends) | |||
| 240 | + | require.NoError(t, d.boundedSendOutputErr(ac, []byte("one-shot effect"))) | |||
| 241 | + | select { | |||
| 242 | + | case frame := <-sends: | |||
| 243 | + | t.Fatalf("parked one-shot effect emitted frame type %d", frame.Type) | |||
| 244 | + | default: | |||
| 245 | + | } | |||
| 246 | + | ||||
| 247 | + | resume := ports.ParkedRouteRequest{RequestID: 2, LeaseID: leaseID, Action: ports.ParkedRouteResume} | |||
| 248 | + | d.handleAttachmentClientFrame(source.attachmentToken(ac, ac.transport()), ports.Frame{Type: ports.MsgParkedRouteRequest, Payload: ports.MarshalParkedRouteRequest(resume)}) | |||
| 249 | + | _ = awaitFrame(t, sends, ports.MsgParkedRouteResponse) | |||
| 250 | + | outputFrame := awaitFrame(t, sends, ports.MsgOutput) | |||
| 251 | + | output, err := ports.UnmarshalOutput(outputFrame.Payload) | |||
| 252 | + | require.NoError(t, err) | |||
| 253 | + | require.True(t, output.Full) | |||
| 254 | + | require.NotContains(t, string(output.Data), "one-shot effect") | |||
| 255 | + | select { | |||
| 256 | + | case frame := <-sends: | |||
| 257 | + | t.Fatalf("parked one-shot effect replayed as frame type %d", frame.Type) | |||
| 258 | + | default: | |||
| 259 | + | } | |||
| 260 | + | } | |||
| 261 | + | ||||
| 262 | + | func TestParkedRouteExpiredLeaseFailsClosed(t *testing.T) { | |||
| 263 | + | d, source, ac, sends, releases := newManualTabSession(t, 1) | |||
| 264 | + | defer releaseAll(releases) | |||
| 265 | + | ac.navigationCapabilities = ports.NavigationCapabilityHomePicker | |||
| 266 | + | ||||
| 267 | + | leaseID := armAndPrepareParkedRoute(t, d, source, ac, sends) | |||
| 268 | + | ||||
| 269 | + | ac.parkedRouteMu.Lock() | |||
| 270 | + | ac.parkedRoute.expiresAt = d.clock.Now() | |||
| 271 | + | ac.parkedRouteMu.Unlock() | |||
| 272 | + | resume := ports.ParkedRouteRequest{RequestID: 2, LeaseID: leaseID, Action: ports.ParkedRouteResume} | |||
| 273 | + | d.handleAttachmentClientFrame(source.attachmentToken(ac, ac.transport()), ports.Frame{Type: ports.MsgParkedRouteRequest, Payload: ports.MarshalParkedRouteRequest(resume)}) | |||
| 274 | + | responseFrame := awaitFrame(t, sends, ports.MsgParkedRouteResponse) | |||
| 275 | + | response, err := ports.UnmarshalParkedRouteResponse(responseFrame.Payload) | |||
| 276 | + | require.NoError(t, err) | |||
| 277 | + | require.Equal(t, ports.ParkedRouteExpired, response.Status) | |||
| 278 | + | require.Same(t, source, ac.currentAttachmentSession()) | |||
| 279 | + | require.False(t, ac.parkedRouteOutput.Load()) | |||
| 280 | + | } | |||
| 281 | + | ||||
| 282 | + | func TestParkedRouteLeaseExpiryClosesExactRetainedTransport(t *testing.T) { | |||
| 283 | + | d, source, ac, _, releases := newManualTabSession(t, 1) | |||
| 284 | + | defer releaseAll(releases) | |||
| 285 | + | expiry := make(chan time.Time, 1) | |||
| 286 | + | timer := portsmocks.NewMockTimer(t) | |||
| 287 | + | timer.EXPECT().C().Return(expiry).Maybe() | |||
| 288 | + | timer.EXPECT().Stop().Return(true).Maybe() | |||
| 289 | + | clock := portsmocks.NewMockClock(t) | |||
| 290 | + | clock.EXPECT().Now().Return(time.Time{}).Maybe() | |||
| 291 | + | clock.EXPECT().NewTimer(mock.Anything).Return(timer).Once() | |||
| 292 | + | d.clock = clock | |||
| 293 | + | transport := &parkedRouteExpiryTransport{sends: make(chan ports.Frame, 8), closed: make(chan struct{})} | |||
| 294 | + | ac.replaceTransport(transport) | |||
| 295 | + | ||||
| 296 | + | _ = armAndPrepareParkedRoute(t, d, source, ac, transport.sends) | |||
| 297 | + | expiry <- time.Time{}.Add(parkedRouteLeaseTTL) | |||
| 298 | + | ||||
| 299 | + | select { | |||
| 300 | + | case <-transport.closed: | |||
| 301 | + | case <-time.After(time.Second): | |||
| 302 | + | t.Fatal("parked transport remained open after its lease expired") | |||
| 303 | + | } | |||
| 304 | + | } | |||
| 305 | + | ||||
| 306 | + | func TestParkedRouteMalformedSwitchRequestsFailClosed(t *testing.T) { | |||
| 307 | + | d, source, ac, sends, releases := newManualTabSession(t, 1) | |||
| 308 | + | defer releaseAll(releases) | |||
| 309 | + | leaseID := ports.ParkedRouteLeaseID{1} | |||
| 310 | + | for _, request := range []ports.ParkedRouteRequest{ | |||
| 311 | + | {RequestID: 1, LeaseID: leaseID, Action: ports.ParkedRouteSwitch}, | |||
| 312 | + | {RequestID: 2, Action: ports.ParkedRouteSwitch, Target: &domain.RemoteSessionTarget{}}, | |||
| 313 | + | {RequestID: 3, LeaseID: leaseID, Action: 99}, | |||
| 314 | + | } { | |||
| 315 | + | d.handleParkedRouteRequest(source.attachmentToken(ac, ac.transport()), request) | |||
| 316 | + | } | |||
| 317 | + | require.Same(t, source, ac.currentAttachmentSession()) | |||
| 318 | + | select { | |||
| 319 | + | case frame := <-sends: | |||
| 320 | + | t.Fatalf("malformed parked route emitted frame type %d", frame.Type) | |||
| 321 | + | default: | |||
| 322 | + | } | |||
| 323 | + | } | |||
| 324 | + | ||||
| 325 | + | func TestParkedRouteResponseFailureDoesNotDrainItsOwnFrameEffect(t *testing.T) { | |||
| 326 | + | d, source, ac, _, releases := newManualTabSession(t, 1) | |||
| 327 | + | defer releaseAll(releases) | |||
| 328 | + | ac.replaceTransport(parkedRouteFailTransport{}) | |||
| 329 | + | token := source.attachmentToken(ac, ac.transport()) | |||
| 330 | + | effect, admitted := ac.beginAttachmentEffect(token) | |||
| 331 | + | require.True(t, admitted) | |||
| 332 | + | token.effect = effect | |||
| 333 | + | ||||
| 334 | + | result := make(chan bool, 1) | |||
| 335 | + | go func() { | |||
| 336 | + | result <- d.respondParkedRoute(token, ports.ParkedRouteResponse{RequestID: 1, Status: ports.ParkedRouteReady}) | |||
| 337 | + | }() | |||
| 338 | + | require.False(t, awaitTestValue(t, result, "parked-route send failure cleanup")) | |||
| 339 | + | d.attachmentCleanupWg.Wait() | |||
| 340 | + | } | |||
| 341 | + | ||||
| 342 | + | func TestParkedRouteDetachClearsLeaseAndSuspension(t *testing.T) { | |||
| 343 | + | d, source, ac, sends, releases := newManualTabSession(t, 1) | |||
| 344 | + | defer releaseAll(releases) | |||
| 345 | + | ac.navigationCapabilities = ports.NavigationCapabilityHomePicker | |||
| 346 | + | ||||
| 347 | + | _ = armAndPrepareParkedRoute(t, d, source, ac, sends) | |||
| 348 | + | require.True(t, ac.parkedRouteOutput.Load()) | |||
| 349 | + | require.NoError(t, d.boundedSendOutputErr(ac, []byte("suppressed side effect"))) | |||
| 350 | + | select { | |||
| 351 | + | case frame := <-sends: | |||
| 352 | + | t.Fatalf("parked side effect emitted frame type %d", frame.Type) | |||
| 353 | + | default: | |||
| 354 | + | } | |||
| 355 | + | ||||
| 356 | + | d.clientGone(source, ac, ac.transport(), true) | |||
| 357 | + | ac.parkedRouteMu.Lock() | |||
| 358 | + | require.Nil(t, ac.parkedRoute) | |||
| 359 | + | ac.parkedRouteMu.Unlock() | |||
| 360 | + | require.False(t, ac.parkedRouteOutput.Load()) | |||
| 361 | + | } | |||
M internal/usecase/daemon/remote_attach.go +22 -2
| @@ -103,11 +103,31 @@ func (d *Daemon) resumeRemoteInactiveSessionLocked(target domain.RemoteSessionTa | |||
| 103 | 103 | } | |
| 104 | 104 | ||
| 105 | 105 | func (d *Daemon) sendNavigationActionForAttachment(token attachmentConnectionToken, action ports.NavigationAction) error { | |
| 106 | - | payload := ports.MarshalNavigationAction(action) | |
| 106 | + | directive := ports.NavigationDirective{Action: action} | |
| 107 | + | armed := false | |
| 108 | + | if action == ports.NavigationOpenHomePicker { | |
| 109 | + | leaseID, err := d.armParkedRoute(token) | |
| 110 | + | if err != nil { | |
| 111 | + | return err | |
| 112 | + | } | |
| 113 | + | directive.LeaseID = leaseID | |
| 114 | + | armed = true | |
| 115 | + | } | |
| 116 | + | rollback := func() { | |
| 117 | + | if armed { | |
| 118 | + | token.ac.clearParkedRoute() | |
| 119 | + | } | |
| 120 | + | } | |
| 121 | + | payload := ports.MarshalNavigationDirective(directive) | |
| 107 | 122 | if payload == nil { | |
| 123 | + | rollback() | |
| 108 | 124 | return errAttachmentTransition | |
| 109 | 125 | } | |
| 110 | - | return token.sendControl(ports.Frame{Type: ports.MsgNavigationAction, Payload: payload}) | |
| 126 | + | if err := token.sendControl(ports.Frame{Type: ports.MsgNavigationAction, Payload: payload}); err != nil { | |
| 127 | + | rollback() | |
| 128 | + | return err | |
| 129 | + | } | |
| 130 | + | return nil | |
| 111 | 131 | } | |
| 112 | 132 | ||
| 113 | 133 | func (d *Daemon) sendRecentRouteNavigationActionForAttachment(token attachmentConnectionToken, action ports.RouteNavigationAction) error { | |
| @@ -103,11 +103,31 @@ func (d *Daemon) resumeRemoteInactiveSessionLocked(target domain.RemoteSessionTa | |||||
| 103 | } | 103 | } | ||
| 104 | 104 | ||||
| 105 | func (d *Daemon) sendNavigationActionForAttachment(token attachmentConnectionToken, action ports.NavigationAction) error { | 105 | func (d *Daemon) sendNavigationActionForAttachment(token attachmentConnectionToken, action ports.NavigationAction) error { | ||
| 106 | - | payload := ports.MarshalNavigationAction(action) | 106 | + | directive := ports.NavigationDirective{Action: action} |
| 107 | + | armed := false | |||
| 108 | + | if action == ports.NavigationOpenHomePicker { | |||
| 109 | + | leaseID, err := d.armParkedRoute(token) | |||
| 110 | + | if err != nil { | |||
| 111 | + | return err | |||
| 112 | + | } | |||
| 113 | + | directive.LeaseID = leaseID | |||
| 114 | + | armed = true | |||
| 115 | + | } | |||
| 116 | + | rollback := func() { | |||
| 117 | + | if armed { | |||
| 118 | + | token.ac.clearParkedRoute() | |||
| 119 | + | } | |||
| 120 | + | } | |||
| 121 | + | payload := ports.MarshalNavigationDirective(directive) | |||
| 107 | if payload == nil { | 122 | if payload == nil { | ||
| 123 | + | rollback() | |||
| 108 | return errAttachmentTransition | 124 | return errAttachmentTransition | ||
| 109 | } | 125 | } | ||
| 110 | - | return token.sendControl(ports.Frame{Type: ports.MsgNavigationAction, Payload: payload}) | 126 | + | if err := token.sendControl(ports.Frame{Type: ports.MsgNavigationAction, Payload: payload}); err != nil { |
| 127 | + | rollback() | |||
| 128 | + | return err | |||
| 129 | + | } | |||
| 130 | + | return nil | |||
| 111 | } | 131 | } | ||
| 112 | 132 | ||||
| 113 | func (d *Daemon) sendRecentRouteNavigationActionForAttachment(token attachmentConnectionToken, action ports.RouteNavigationAction) error { | 133 | func (d *Daemon) sendRecentRouteNavigationActionForAttachment(token attachmentConnectionToken, action ports.RouteNavigationAction) error { | ||
M internal/usecase/daemon/remote_picker_test.go +10 -2
| @@ -855,6 +855,9 @@ func TestNavigationActionHandoffSendsBoundedAction(t *testing.T) { | |||
| 855 | 855 | t.Run(tt.name, func(t *testing.T) { | |
| 856 | 856 | d := newRemotePickerDaemon(nil) | |
| 857 | 857 | sess, ac, sends := addRemoteRefreshPickerOwner(t, d, "local") | |
| 858 | + | if tt.action == ports.NavigationOpenHomePicker { | |
| 859 | + | ac.navigationCapabilities = ports.NavigationCapabilityHomePicker | |
| 860 | + | } | |
| 858 | 861 | token := sess.attachmentToken(ac, ac.transport()) | |
| 859 | 862 | effect, admitted := ac.beginAttachmentEffect(token) | |
| 860 | 863 | require.True(t, admitted) | |
| @@ -862,9 +865,14 @@ func TestNavigationActionHandoffSendsBoundedAction(t *testing.T) { | |||
| 862 | 865 | require.NoError(t, d.sendNavigationActionForAttachment(token, tt.action)) | |
| 863 | 866 | frame := receiveRemotePicker(t, sends, "navigation action") | |
| 864 | 867 | require.Equal(t, ports.MsgNavigationAction, frame.Type) | |
| 865 | - | got, err := ports.UnmarshalNavigationAction(frame.Payload) | |
| 868 | + | directive, err := ports.UnmarshalNavigationDirective(frame.Payload) | |
| 866 | 869 | require.NoError(t, err) | |
| 867 | - | require.Equal(t, tt.action, got) | |
| 870 | + | require.Equal(t, tt.action, directive.Action) | |
| 871 | + | if tt.action == ports.NavigationOpenHomePicker { | |
| 872 | + | require.False(t, directive.LeaseID.IsZero()) | |
| 873 | + | } else { | |
| 874 | + | require.True(t, directive.LeaseID.IsZero()) | |
| 875 | + | } | |
| 868 | 876 | }) | |
| 869 | 877 | } | |
| 870 | 878 | } | |
| @@ -855,6 +855,9 @@ func TestNavigationActionHandoffSendsBoundedAction(t *testing.T) { | |||||
| 855 | t.Run(tt.name, func(t *testing.T) { | 855 | t.Run(tt.name, func(t *testing.T) { | ||
| 856 | d := newRemotePickerDaemon(nil) | 856 | d := newRemotePickerDaemon(nil) | ||
| 857 | sess, ac, sends := addRemoteRefreshPickerOwner(t, d, "local") | 857 | sess, ac, sends := addRemoteRefreshPickerOwner(t, d, "local") | ||
| 858 | + | if tt.action == ports.NavigationOpenHomePicker { | |||
| 859 | + | ac.navigationCapabilities = ports.NavigationCapabilityHomePicker | |||
| 860 | + | } | |||
| 858 | token := sess.attachmentToken(ac, ac.transport()) | 861 | token := sess.attachmentToken(ac, ac.transport()) | ||
| 859 | effect, admitted := ac.beginAttachmentEffect(token) | 862 | effect, admitted := ac.beginAttachmentEffect(token) | ||
| 860 | require.True(t, admitted) | 863 | require.True(t, admitted) | ||
| @@ -862,9 +865,14 @@ func TestNavigationActionHandoffSendsBoundedAction(t *testing.T) { | |||||
| 862 | require.NoError(t, d.sendNavigationActionForAttachment(token, tt.action)) | 865 | require.NoError(t, d.sendNavigationActionForAttachment(token, tt.action)) | ||
| 863 | frame := receiveRemotePicker(t, sends, "navigation action") | 866 | frame := receiveRemotePicker(t, sends, "navigation action") | ||
| 864 | require.Equal(t, ports.MsgNavigationAction, frame.Type) | 867 | require.Equal(t, ports.MsgNavigationAction, frame.Type) | ||
| 865 | - | got, err := ports.UnmarshalNavigationAction(frame.Payload) | 868 | + | directive, err := ports.UnmarshalNavigationDirective(frame.Payload) |
| 866 | require.NoError(t, err) | 869 | require.NoError(t, err) | ||
| 867 | - | require.Equal(t, tt.action, got) | 870 | + | require.Equal(t, tt.action, directive.Action) |
| 871 | + | if tt.action == ports.NavigationOpenHomePicker { | |||
| 872 | + | require.False(t, directive.LeaseID.IsZero()) | |||
| 873 | + | } else { | |||
| 874 | + | require.True(t, directive.LeaseID.IsZero()) | |||
| 875 | + | } | |||
| 868 | }) | 876 | }) | ||
| 869 | } | 877 | } | ||
| 870 | } | 878 | } | ||
M internal/usecase/daemon/render.go +8 -2
| @@ -329,10 +329,13 @@ func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *at | |||
| 329 | 329 | entry.core().mu.Lock() | |
| 330 | 330 | _, owned := entry.core().attachments[ac] | |
| 331 | 331 | entry.core().mu.Unlock() | |
| 332 | - | if !owned || ac.currentAttachmentSession() != entry { | |
| 332 | + | if !owned || ac.currentAttachmentSession() != entry || ac.parkedRouteOutput.Load() { | |
| 333 | 333 | ac.sendMu.Unlock() | |
| 334 | 334 | return paintRejected | |
| 335 | 335 | } | |
| 336 | + | if ac.parkedRouteFullPending.Load() { | |
| 337 | + | reset = true | |
| 338 | + | } | |
| 336 | 339 | if lease != nil { | |
| 337 | 340 | rc := attachmentRenderCoordinator(entry) | |
| 338 | 341 | if rc == nil || lease.attachment != ac || !rc.leaseCurrent(lease, true) { | |
| @@ -455,7 +458,10 @@ func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *at | |||
| 455 | 458 | if ac.renderStages.compose != nil { | |
| 456 | 459 | ac.renderStages.compose() | |
| 457 | 460 | } | |
| 458 | - | d.emitFrame(entry, ac, state, composed, &marks) | |
| 461 | + | full := state.reset | |
| 462 | + | if d.emitFrame(entry, ac, state, composed, &marks) && full { | |
| 463 | + | ac.parkedRouteFullPending.Store(false) | |
| 464 | + | } | |
| 459 | 465 | return paintEmitted | |
| 460 | 466 | } | |
| 461 | 467 | ||
| @@ -329,10 +329,13 @@ func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *at | |||||
| 329 | entry.core().mu.Lock() | 329 | entry.core().mu.Lock() | ||
| 330 | _, owned := entry.core().attachments[ac] | 330 | _, owned := entry.core().attachments[ac] | ||
| 331 | entry.core().mu.Unlock() | 331 | entry.core().mu.Unlock() | ||
| 332 | - | if !owned || ac.currentAttachmentSession() != entry { | 332 | + | if !owned || ac.currentAttachmentSession() != entry || ac.parkedRouteOutput.Load() { |
| 333 | ac.sendMu.Unlock() | 333 | ac.sendMu.Unlock() | ||
| 334 | return paintRejected | 334 | return paintRejected | ||
| 335 | } | 335 | } | ||
| 336 | + | if ac.parkedRouteFullPending.Load() { | |||
| 337 | + | reset = true | |||
| 338 | + | } | |||
| 336 | if lease != nil { | 339 | if lease != nil { | ||
| 337 | rc := attachmentRenderCoordinator(entry) | 340 | rc := attachmentRenderCoordinator(entry) | ||
| 338 | if rc == nil || lease.attachment != ac || !rc.leaseCurrent(lease, true) { | 341 | if rc == nil || lease.attachment != ac || !rc.leaseCurrent(lease, true) { | ||
| @@ -455,7 +458,10 @@ func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *at | |||||
| 455 | if ac.renderStages.compose != nil { | 458 | if ac.renderStages.compose != nil { | ||
| 456 | ac.renderStages.compose() | 459 | ac.renderStages.compose() | ||
| 457 | } | 460 | } | ||
| 458 | - | d.emitFrame(entry, ac, state, composed, &marks) | 461 | + | full := state.reset |
| 462 | + | if d.emitFrame(entry, ac, state, composed, &marks) && full { | |||
| 463 | + | ac.parkedRouteFullPending.Store(false) | |||
| 464 | + | } | |||
| 459 | return paintEmitted | 465 | return paintEmitted | ||
| 460 | } | 466 | } | ||
| 461 | 467 | ||||