fix(daemon): fence remote link reconnection
- commit
- afa0cb541ab9ff0398b50feba6ce764e5462690f
- author
- brice · · 2026-08-06 20:12 UTC
- committer
- brice · · 2026-08-06 20:14 UTC
- parent
- c48b08e
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +363 -21
M internal/usecase/daemon/remote_link.go +146 -20
| @@ -27,6 +27,12 @@ type remoteViewConstruction struct { | |||
| 27 | 27 | err error | |
| 28 | 28 | } | |
| 29 | 29 | ||
| 30 | + | type remoteViewLinkSnapshot struct { | |
| 31 | + | view *remoteView | |
| 32 | + | link *remoteLink | |
| 33 | + | generation uint64 | |
| 34 | + | } | |
| 35 | + | ||
| 30 | 36 | // remoteLink is one exact transport generation for a remote view. sendMu | |
| 31 | 37 | // orders every outbound remote frame; view.mu guards whether this exact link | |
| 32 | 38 | // remains current. Neither lock is held while transport I/O runs. | |
| @@ -92,9 +98,10 @@ func (s remoteOutputState) next(output ports.Output) (remoteOutputState, bool) { | |||
| 92 | 98 | return remoteOutputState{epoch: output.Epoch, state: output.New, viewRevision: output.ViewRevision, initialized: true}, true | |
| 93 | 99 | } | |
| 94 | 100 | ||
| 95 | - | // openRemoteView returns the exact warm view for target or elects one | |
| 96 | - | // constructor. The reservation is protected only by Daemon.mu; the dial and | |
| 97 | - | // all handshake I/O happen after every architecture lock is released. | |
| 101 | + | // openRemoteView returns the exact healthy warm view for target or elects one | |
| 102 | + | // constructor. An unhealthy registry hit is a reconnect candidate, never a | |
| 103 | + | // reusable result. The reservation is protected only by Daemon.mu; the dial | |
| 104 | + | // and all handshake I/O happen after every architecture lock is released. | |
| 98 | 105 | func (d *Daemon) openRemoteView(ctx context.Context, target domain.RemoteSessionTarget, size domain.Size) (*remoteView, error) { | |
| 99 | 106 | if d == nil { | |
| 100 | 107 | return nil, errors.New("remote view: nil daemon") | |
| @@ -121,7 +128,8 @@ func (d *Daemon) openRemoteView(ctx context.Context, target domain.RemoteSession | |||
| 121 | 128 | return nil, err | |
| 122 | 129 | } | |
| 123 | 130 | d.mu.Lock() | |
| 124 | - | if existing := d.remoteViewByKeyLocked(key); existing != nil { | |
| 131 | + | existing, observed, reusable := d.remoteViewLinkByKeyLocked(key) | |
| 132 | + | if reusable { | |
| 125 | 133 | d.mu.Unlock() | |
| 126 | 134 | return existing, nil | |
| 127 | 135 | } | |
| @@ -141,17 +149,17 @@ func (d *Daemon) openRemoteView(ctx context.Context, target domain.RemoteSession | |||
| 141 | 149 | return nil, err | |
| 142 | 150 | } | |
| 143 | 151 | d.mu.Lock() | |
| 144 | - | winner := d.remoteViewByKeyLocked(key) | |
| 152 | + | winner, _, healthy := d.remoteViewLinkByKeyLocked(key) | |
| 145 | 153 | constructionErr := construction.err | |
| 146 | 154 | d.mu.Unlock() | |
| 147 | - | if winner != nil { | |
| 155 | + | if healthy { | |
| 148 | 156 | return winner, nil | |
| 149 | 157 | } | |
| 150 | 158 | if constructionErr != nil && !errors.Is(constructionErr, context.Canceled) && !errors.Is(constructionErr, context.DeadlineExceeded) { | |
| 151 | 159 | return nil, constructionErr | |
| 152 | 160 | } | |
| 153 | - | // A canceled candidate never poisons an independent waiter. Re-elect | |
| 154 | - | // if the caller still wants this exact remote view. | |
| 161 | + | // A canceled candidate never poisons an independent waiter. A link | |
| 162 | + | // that failed immediately after publication is likewise not reusable. | |
| 155 | 163 | continue | |
| 156 | 164 | } | |
| 157 | 165 | ||
| @@ -170,21 +178,55 @@ func (d *Daemon) openRemoteView(ctx context.Context, target domain.RemoteSession | |||
| 170 | 178 | cancel() | |
| 171 | 179 | ||
| 172 | 180 | var winner *remoteView | |
| 181 | + | var replaced *remoteLink | |
| 182 | + | var repaint []*attachedClient | |
| 173 | 183 | publishErr := constructionErr | |
| 174 | 184 | d.mu.Lock() | |
| 175 | 185 | if publishErr == nil { | |
| 176 | - | if d.closing { | |
| 186 | + | switch current := d.remoteViewByKeyLocked(key); { | |
| 187 | + | case d.closing: | |
| 177 | 188 | publishErr = errors.New("remote view: daemon is shutting down") | |
| 178 | - | } else if existing := d.remoteViewByKeyLocked(key); existing != nil { | |
| 179 | - | winner = existing | |
| 180 | - | } else if err := d.registerRemoteViewLocked(candidate); err != nil { | |
| 181 | - | publishErr = fmt.Errorf("remote view: publish candidate: %w", err) | |
| 182 | - | } else { | |
| 183 | - | winner = candidate | |
| 184 | - | // Starting the receiver while d.mu still protects publication keeps | |
| 185 | - | // shutdown from observing a registry-visible link with no worker | |
| 186 | - | // to join. | |
| 187 | - | d.startRemoteLink(candidate.link) | |
| 189 | + | case current == nil && observed.view != nil: | |
| 190 | + | publishErr = errRemoteViewStale | |
| 191 | + | case current == nil: | |
| 192 | + | if err := d.registerRemoteViewLocked(candidate); err != nil { | |
| 193 | + | publishErr = fmt.Errorf("remote view: publish candidate: %w", err) | |
| 194 | + | } else { | |
| 195 | + | winner = candidate | |
| 196 | + | // Starting the receiver while d.mu still protects publication | |
| 197 | + | // keeps shutdown from observing a registry-visible link with no | |
| 198 | + | // worker to join. | |
| 199 | + | d.startRemoteLink(candidate.link) | |
| 200 | + | } | |
| 201 | + | case current != observed.view: | |
| 202 | + | current.mu.Lock() | |
| 203 | + | if remoteViewLinkReusableLocked(current) { | |
| 204 | + | winner = current | |
| 205 | + | } else { | |
| 206 | + | publishErr = errRemoteViewStale | |
| 207 | + | } | |
| 208 | + | current.mu.Unlock() | |
| 209 | + | default: | |
| 210 | + | current.mu.Lock() | |
| 211 | + | if remoteViewLinkReusableLocked(current) { | |
| 212 | + | winner = current | |
| 213 | + | } else if current.closed || current.link != observed.link || current.linkGeneration != observed.generation { | |
| 214 | + | publishErr = errRemoteViewStale | |
| 215 | + | } else { | |
| 216 | + | var installed bool | |
| 217 | + | replaced, installed = installRemoteViewCandidateLocked(current, candidate) | |
| 218 | + | if !installed { | |
| 219 | + | publishErr = errors.New("remote view: candidate link is unavailable") | |
| 220 | + | } else { | |
| 221 | + | winner = current | |
| 222 | + | repaint = make([]*attachedClient, 0, len(current.attachments)) | |
| 223 | + | for attachment := range current.attachments { | |
| 224 | + | repaint = append(repaint, attachment) | |
| 225 | + | } | |
| 226 | + | d.startRemoteLink(current.link) | |
| 227 | + | } | |
| 228 | + | } | |
| 229 | + | current.mu.Unlock() | |
| 188 | 230 | } | |
| 189 | 231 | } | |
| 190 | 232 | construction.err = publishErr | |
| @@ -197,13 +239,92 @@ func (d *Daemon) openRemoteView(ctx context.Context, target domain.RemoteSession | |||
| 197 | 239 | if candidate != nil && candidate != winner { | |
| 198 | 240 | d.stopUnpublishedRemoteView(candidate) | |
| 199 | 241 | } | |
| 242 | + | if replaced != nil { | |
| 243 | + | stopAndJoinRemoteLink(replaced) | |
| 244 | + | } | |
| 245 | + | for _, attachment := range repaint { | |
| 246 | + | token := attachmentOwnerToken(winner, attachment, attachment.transport()) | |
| 247 | + | if token.ac != nil { | |
| 248 | + | go d.paintRemoteView(winner, attachment, false, token) | |
| 249 | + | } | |
| 250 | + | } | |
| 200 | 251 | if winner != nil { | |
| 201 | - | return winner, nil | |
| 252 | + | d.mu.Lock() | |
| 253 | + | current, _, healthy := d.remoteViewLinkByKeyLocked(key) | |
| 254 | + | d.mu.Unlock() | |
| 255 | + | if healthy && current == winner { | |
| 256 | + | return winner, nil | |
| 257 | + | } | |
| 258 | + | continue | |
| 202 | 259 | } | |
| 203 | 260 | return nil, publishErr | |
| 204 | 261 | } | |
| 205 | 262 | } | |
| 206 | 263 | ||
| 264 | + | // remoteViewLinkByKeyLocked snapshots health and replacement authority under | |
| 265 | + | // the registry-to-view lock order. Caller holds d.mu. | |
| 266 | + | func (d *Daemon) remoteViewLinkByKeyLocked(key remoteViewKey) (*remoteView, remoteViewLinkSnapshot, bool) { | |
| 267 | + | view := d.remoteViewByKeyLocked(key) | |
| 268 | + | if view == nil { | |
| 269 | + | return nil, remoteViewLinkSnapshot{}, false | |
| 270 | + | } | |
| 271 | + | view.mu.Lock() | |
| 272 | + | snapshot := remoteViewLinkSnapshot{view: view, link: view.link, generation: view.linkGeneration} | |
| 273 | + | reusable := remoteViewLinkReusableLocked(view) | |
| 274 | + | view.mu.Unlock() | |
| 275 | + | return view, snapshot, reusable | |
| 276 | + | } | |
| 277 | + | ||
| 278 | + | func remoteViewLinkReusableLocked(view *remoteView) bool { | |
| 279 | + | if view == nil || view.closed || view.link == nil { | |
| 280 | + | return false | |
| 281 | + | } | |
| 282 | + | link := view.link | |
| 283 | + | return link.view == view && link.generation == view.linkGeneration && link.transport != nil && link.active | |
| 284 | + | } | |
| 285 | + | ||
| 286 | + | // installRemoteViewCandidateLocked transfers one fully handshaken candidate | |
| 287 | + | // into the exact observed view generation. Attachment membership and warm | |
| 288 | + | // retention stay on the stable view; content becomes visible only with the new | |
| 289 | + | // validated link publication. Caller holds the existing view's mutex. | |
| 290 | + | func installRemoteViewCandidateLocked(view, candidate *remoteView) (*remoteLink, bool) { | |
| 291 | + | if view == nil || candidate == nil || candidate.link == nil || candidate.screen == nil { | |
| 292 | + | return nil, false | |
| 293 | + | } | |
| 294 | + | link := candidate.link | |
| 295 | + | replaced := view.link | |
| 296 | + | view.linkGeneration++ | |
| 297 | + | link.view = view | |
| 298 | + | link.generation = view.linkGeneration | |
| 299 | + | link.active = true | |
| 300 | + | view.link = link | |
| 301 | + | view.screen = candidate.screen | |
| 302 | + | view.metadata = cloneRemoteSessionMeta(candidate.metadata) | |
| 303 | + | view.displayOrigin = candidate.displayOrigin | |
| 304 | + | view.output = candidate.output | |
| 305 | + | view.resetRequested = candidate.resetRequested | |
| 306 | + | if replaced != nil { | |
| 307 | + | replaced.active = false | |
| 308 | + | } | |
| 309 | + | candidate.link = nil | |
| 310 | + | return replaced, true | |
| 311 | + | } | |
| 312 | + | ||
| 313 | + | func stopAndJoinRemoteLink(link *remoteLink) { | |
| 314 | + | if link == nil { | |
| 315 | + | return | |
| 316 | + | } | |
| 317 | + | if link.cancel != nil { | |
| 318 | + | link.cancel() | |
| 319 | + | } | |
| 320 | + | if link.transport != nil { | |
| 321 | + | _ = link.transport.Close() | |
| 322 | + | } | |
| 323 | + | if link.done != nil { | |
| 324 | + | <-link.done | |
| 325 | + | } | |
| 326 | + | } | |
| 327 | + | ||
| 207 | 328 | // constructRemoteView builds an unexposed candidate. Its caller owns a | |
| 208 | 329 | // constructor reservation but holds no daemon, session, or view lock. | |
| 209 | 330 | func (d *Daemon) constructRemoteView(ctx context.Context, target domain.RemoteSessionTarget, size domain.Size) (_ *remoteView, err error) { | |
| @@ -618,6 +739,11 @@ func (link *remoteLink) send(frame ports.Frame) error { | |||
| 618 | 739 | err := transport.Send(frame) | |
| 619 | 740 | link.sendMu.Unlock() | |
| 620 | 741 | if err != nil { | |
| 742 | + | view.mu.Lock() | |
| 743 | + | if view.link == link && view.linkGeneration == link.generation { | |
| 744 | + | link.active = false | |
| 745 | + | } | |
| 746 | + | view.mu.Unlock() | |
| 621 | 747 | link.cancel() | |
| 622 | 748 | _ = transport.Close() | |
| 623 | 749 | } | |
| @@ -27,6 +27,12 @@ type remoteViewConstruction struct { | |||||
| 27 | err error | 27 | err error | ||
| 28 | } | 28 | } | ||
| 29 | 29 | ||||
| 30 | + | type remoteViewLinkSnapshot struct { | |||
| 31 | + | view *remoteView | |||
| 32 | + | link *remoteLink | |||
| 33 | + | generation uint64 | |||
| 34 | + | } | |||
| 35 | + | ||||
| 30 | // remoteLink is one exact transport generation for a remote view. sendMu | 36 | // remoteLink is one exact transport generation for a remote view. sendMu | ||
| 31 | // orders every outbound remote frame; view.mu guards whether this exact link | 37 | // orders every outbound remote frame; view.mu guards whether this exact link | ||
| 32 | // remains current. Neither lock is held while transport I/O runs. | 38 | // remains current. Neither lock is held while transport I/O runs. | ||
| @@ -92,9 +98,10 @@ func (s remoteOutputState) next(output ports.Output) (remoteOutputState, bool) { | |||||
| 92 | return remoteOutputState{epoch: output.Epoch, state: output.New, viewRevision: output.ViewRevision, initialized: true}, true | 98 | return remoteOutputState{epoch: output.Epoch, state: output.New, viewRevision: output.ViewRevision, initialized: true}, true | ||
| 93 | } | 99 | } | ||
| 94 | 100 | ||||
| 95 | - | // openRemoteView returns the exact warm view for target or elects one | 101 | + | // openRemoteView returns the exact healthy warm view for target or elects one |
| 96 | - | // constructor. The reservation is protected only by Daemon.mu; the dial and | 102 | + | // constructor. An unhealthy registry hit is a reconnect candidate, never a |
| 97 | - | // all handshake I/O happen after every architecture lock is released. | 103 | + | // reusable result. The reservation is protected only by Daemon.mu; the dial |
| 104 | + | // and all handshake I/O happen after every architecture lock is released. | |||
| 98 | func (d *Daemon) openRemoteView(ctx context.Context, target domain.RemoteSessionTarget, size domain.Size) (*remoteView, error) { | 105 | func (d *Daemon) openRemoteView(ctx context.Context, target domain.RemoteSessionTarget, size domain.Size) (*remoteView, error) { | ||
| 99 | if d == nil { | 106 | if d == nil { | ||
| 100 | return nil, errors.New("remote view: nil daemon") | 107 | return nil, errors.New("remote view: nil daemon") | ||
| @@ -121,7 +128,8 @@ func (d *Daemon) openRemoteView(ctx context.Context, target domain.RemoteSession | |||||
| 121 | return nil, err | 128 | return nil, err | ||
| 122 | } | 129 | } | ||
| 123 | d.mu.Lock() | 130 | d.mu.Lock() | ||
| 124 | - | if existing := d.remoteViewByKeyLocked(key); existing != nil { | 131 | + | existing, observed, reusable := d.remoteViewLinkByKeyLocked(key) |
| 132 | + | if reusable { | |||
| 125 | d.mu.Unlock() | 133 | d.mu.Unlock() | ||
| 126 | return existing, nil | 134 | return existing, nil | ||
| 127 | } | 135 | } | ||
| @@ -141,17 +149,17 @@ func (d *Daemon) openRemoteView(ctx context.Context, target domain.RemoteSession | |||||
| 141 | return nil, err | 149 | return nil, err | ||
| 142 | } | 150 | } | ||
| 143 | d.mu.Lock() | 151 | d.mu.Lock() | ||
| 144 | - | winner := d.remoteViewByKeyLocked(key) | 152 | + | winner, _, healthy := d.remoteViewLinkByKeyLocked(key) |
| 145 | constructionErr := construction.err | 153 | constructionErr := construction.err | ||
| 146 | d.mu.Unlock() | 154 | d.mu.Unlock() | ||
| 147 | - | if winner != nil { | 155 | + | if healthy { |
| 148 | return winner, nil | 156 | return winner, nil | ||
| 149 | } | 157 | } | ||
| 150 | if constructionErr != nil && !errors.Is(constructionErr, context.Canceled) && !errors.Is(constructionErr, context.DeadlineExceeded) { | 158 | if constructionErr != nil && !errors.Is(constructionErr, context.Canceled) && !errors.Is(constructionErr, context.DeadlineExceeded) { | ||
| 151 | return nil, constructionErr | 159 | return nil, constructionErr | ||
| 152 | } | 160 | } | ||
| 153 | - | // A canceled candidate never poisons an independent waiter. Re-elect | 161 | + | // A canceled candidate never poisons an independent waiter. A link |
| 154 | - | // if the caller still wants this exact remote view. | 162 | + | // that failed immediately after publication is likewise not reusable. |
| 155 | continue | 163 | continue | ||
| 156 | } | 164 | } | ||
| 157 | 165 | ||||
| @@ -170,21 +178,55 @@ func (d *Daemon) openRemoteView(ctx context.Context, target domain.RemoteSession | |||||
| 170 | cancel() | 178 | cancel() | ||
| 171 | 179 | ||||
| 172 | var winner *remoteView | 180 | var winner *remoteView | ||
| 181 | + | var replaced *remoteLink | |||
| 182 | + | var repaint []*attachedClient | |||
| 173 | publishErr := constructionErr | 183 | publishErr := constructionErr | ||
| 174 | d.mu.Lock() | 184 | d.mu.Lock() | ||
| 175 | if publishErr == nil { | 185 | if publishErr == nil { | ||
| 176 | - | if d.closing { | 186 | + | switch current := d.remoteViewByKeyLocked(key); { |
| 187 | + | case d.closing: | |||
| 177 | publishErr = errors.New("remote view: daemon is shutting down") | 188 | publishErr = errors.New("remote view: daemon is shutting down") | ||
| 178 | - | } else if existing := d.remoteViewByKeyLocked(key); existing != nil { | 189 | + | case current == nil && observed.view != nil: |
| 179 | - | winner = existing | 190 | + | publishErr = errRemoteViewStale |
| 180 | - | } else if err := d.registerRemoteViewLocked(candidate); err != nil { | 191 | + | case current == nil: |
| 181 | - | publishErr = fmt.Errorf("remote view: publish candidate: %w", err) | 192 | + | if err := d.registerRemoteViewLocked(candidate); err != nil { |
| 182 | - | } else { | 193 | + | publishErr = fmt.Errorf("remote view: publish candidate: %w", err) |
| 183 | - | winner = candidate | 194 | + | } else { |
| 184 | - | // Starting the receiver while d.mu still protects publication keeps | 195 | + | winner = candidate |
| 185 | - | // shutdown from observing a registry-visible link with no worker | 196 | + | // Starting the receiver while d.mu still protects publication |
| 186 | - | // to join. | 197 | + | // keeps shutdown from observing a registry-visible link with no |
| 187 | - | d.startRemoteLink(candidate.link) | 198 | + | // worker to join. |
| 199 | + | d.startRemoteLink(candidate.link) | |||
| 200 | + | } | |||
| 201 | + | case current != observed.view: | |||
| 202 | + | current.mu.Lock() | |||
| 203 | + | if remoteViewLinkReusableLocked(current) { | |||
| 204 | + | winner = current | |||
| 205 | + | } else { | |||
| 206 | + | publishErr = errRemoteViewStale | |||
| 207 | + | } | |||
| 208 | + | current.mu.Unlock() | |||
| 209 | + | default: | |||
| 210 | + | current.mu.Lock() | |||
| 211 | + | if remoteViewLinkReusableLocked(current) { | |||
| 212 | + | winner = current | |||
| 213 | + | } else if current.closed || current.link != observed.link || current.linkGeneration != observed.generation { | |||
| 214 | + | publishErr = errRemoteViewStale | |||
| 215 | + | } else { | |||
| 216 | + | var installed bool | |||
| 217 | + | replaced, installed = installRemoteViewCandidateLocked(current, candidate) | |||
| 218 | + | if !installed { | |||
| 219 | + | publishErr = errors.New("remote view: candidate link is unavailable") | |||
| 220 | + | } else { | |||
| 221 | + | winner = current | |||
| 222 | + | repaint = make([]*attachedClient, 0, len(current.attachments)) | |||
| 223 | + | for attachment := range current.attachments { | |||
| 224 | + | repaint = append(repaint, attachment) | |||
| 225 | + | } | |||
| 226 | + | d.startRemoteLink(current.link) | |||
| 227 | + | } | |||
| 228 | + | } | |||
| 229 | + | current.mu.Unlock() | |||
| 188 | } | 230 | } | ||
| 189 | } | 231 | } | ||
| 190 | construction.err = publishErr | 232 | construction.err = publishErr | ||
| @@ -197,13 +239,92 @@ func (d *Daemon) openRemoteView(ctx context.Context, target domain.RemoteSession | |||||
| 197 | if candidate != nil && candidate != winner { | 239 | if candidate != nil && candidate != winner { | ||
| 198 | d.stopUnpublishedRemoteView(candidate) | 240 | d.stopUnpublishedRemoteView(candidate) | ||
| 199 | } | 241 | } | ||
| 242 | + | if replaced != nil { | |||
| 243 | + | stopAndJoinRemoteLink(replaced) | |||
| 244 | + | } | |||
| 245 | + | for _, attachment := range repaint { | |||
| 246 | + | token := attachmentOwnerToken(winner, attachment, attachment.transport()) | |||
| 247 | + | if token.ac != nil { | |||
| 248 | + | go d.paintRemoteView(winner, attachment, false, token) | |||
| 249 | + | } | |||
| 250 | + | } | |||
| 200 | if winner != nil { | 251 | if winner != nil { | ||
| 201 | - | return winner, nil | 252 | + | d.mu.Lock() |
| 253 | + | current, _, healthy := d.remoteViewLinkByKeyLocked(key) | |||
| 254 | + | d.mu.Unlock() | |||
| 255 | + | if healthy && current == winner { | |||
| 256 | + | return winner, nil | |||
| 257 | + | } | |||
| 258 | + | continue | |||
| 202 | } | 259 | } | ||
| 203 | return nil, publishErr | 260 | return nil, publishErr | ||
| 204 | } | 261 | } | ||
| 205 | } | 262 | } | ||
| 206 | 263 | ||||
| 264 | + | // remoteViewLinkByKeyLocked snapshots health and replacement authority under | |||
| 265 | + | // the registry-to-view lock order. Caller holds d.mu. | |||
| 266 | + | func (d *Daemon) remoteViewLinkByKeyLocked(key remoteViewKey) (*remoteView, remoteViewLinkSnapshot, bool) { | |||
| 267 | + | view := d.remoteViewByKeyLocked(key) | |||
| 268 | + | if view == nil { | |||
| 269 | + | return nil, remoteViewLinkSnapshot{}, false | |||
| 270 | + | } | |||
| 271 | + | view.mu.Lock() | |||
| 272 | + | snapshot := remoteViewLinkSnapshot{view: view, link: view.link, generation: view.linkGeneration} | |||
| 273 | + | reusable := remoteViewLinkReusableLocked(view) | |||
| 274 | + | view.mu.Unlock() | |||
| 275 | + | return view, snapshot, reusable | |||
| 276 | + | } | |||
| 277 | + | ||||
| 278 | + | func remoteViewLinkReusableLocked(view *remoteView) bool { | |||
| 279 | + | if view == nil || view.closed || view.link == nil { | |||
| 280 | + | return false | |||
| 281 | + | } | |||
| 282 | + | link := view.link | |||
| 283 | + | return link.view == view && link.generation == view.linkGeneration && link.transport != nil && link.active | |||
| 284 | + | } | |||
| 285 | + | ||||
| 286 | + | // installRemoteViewCandidateLocked transfers one fully handshaken candidate | |||
| 287 | + | // into the exact observed view generation. Attachment membership and warm | |||
| 288 | + | // retention stay on the stable view; content becomes visible only with the new | |||
| 289 | + | // validated link publication. Caller holds the existing view's mutex. | |||
| 290 | + | func installRemoteViewCandidateLocked(view, candidate *remoteView) (*remoteLink, bool) { | |||
| 291 | + | if view == nil || candidate == nil || candidate.link == nil || candidate.screen == nil { | |||
| 292 | + | return nil, false | |||
| 293 | + | } | |||
| 294 | + | link := candidate.link | |||
| 295 | + | replaced := view.link | |||
| 296 | + | view.linkGeneration++ | |||
| 297 | + | link.view = view | |||
| 298 | + | link.generation = view.linkGeneration | |||
| 299 | + | link.active = true | |||
| 300 | + | view.link = link | |||
| 301 | + | view.screen = candidate.screen | |||
| 302 | + | view.metadata = cloneRemoteSessionMeta(candidate.metadata) | |||
| 303 | + | view.displayOrigin = candidate.displayOrigin | |||
| 304 | + | view.output = candidate.output | |||
| 305 | + | view.resetRequested = candidate.resetRequested | |||
| 306 | + | if replaced != nil { | |||
| 307 | + | replaced.active = false | |||
| 308 | + | } | |||
| 309 | + | candidate.link = nil | |||
| 310 | + | return replaced, true | |||
| 311 | + | } | |||
| 312 | + | ||||
| 313 | + | func stopAndJoinRemoteLink(link *remoteLink) { | |||
| 314 | + | if link == nil { | |||
| 315 | + | return | |||
| 316 | + | } | |||
| 317 | + | if link.cancel != nil { | |||
| 318 | + | link.cancel() | |||
| 319 | + | } | |||
| 320 | + | if link.transport != nil { | |||
| 321 | + | _ = link.transport.Close() | |||
| 322 | + | } | |||
| 323 | + | if link.done != nil { | |||
| 324 | + | <-link.done | |||
| 325 | + | } | |||
| 326 | + | } | |||
| 327 | + | ||||
| 207 | // constructRemoteView builds an unexposed candidate. Its caller owns a | 328 | // constructRemoteView builds an unexposed candidate. Its caller owns a | ||
| 208 | // constructor reservation but holds no daemon, session, or view lock. | 329 | // constructor reservation but holds no daemon, session, or view lock. | ||
| 209 | func (d *Daemon) constructRemoteView(ctx context.Context, target domain.RemoteSessionTarget, size domain.Size) (_ *remoteView, err error) { | 330 | func (d *Daemon) constructRemoteView(ctx context.Context, target domain.RemoteSessionTarget, size domain.Size) (_ *remoteView, err error) { | ||
| @@ -618,6 +739,11 @@ func (link *remoteLink) send(frame ports.Frame) error { | |||||
| 618 | err := transport.Send(frame) | 739 | err := transport.Send(frame) | ||
| 619 | link.sendMu.Unlock() | 740 | link.sendMu.Unlock() | ||
| 620 | if err != nil { | 741 | if err != nil { | ||
| 742 | + | view.mu.Lock() | |||
| 743 | + | if view.link == link && view.linkGeneration == link.generation { | |||
| 744 | + | link.active = false | |||
| 745 | + | } | |||
| 746 | + | view.mu.Unlock() | |||
| 621 | link.cancel() | 747 | link.cancel() | ||
| 622 | _ = transport.Close() | 748 | _ = transport.Close() | ||
| 623 | } | 749 | } | ||
M internal/usecase/daemon/remote_link_test.go +217 -1
| @@ -88,6 +88,22 @@ func (f *remoteLinkTestFactory) DialerForRemote(_ string, _ string, _ ports.Remo | |||
| 88 | 88 | return f.dialer, nil | |
| 89 | 89 | } | |
| 90 | 90 | ||
| 91 | + | type remoteLinkSequenceFactory struct { | |
| 92 | + | mu sync.Mutex | |
| 93 | + | dialers []ports.Dialer | |
| 94 | + | calls atomic.Int32 | |
| 95 | + | } | |
| 96 | + | ||
| 97 | + | func (f *remoteLinkSequenceFactory) DialerForRemote(_ string, _ string, _ ports.RemoteTransportMode, _ *slog.Logger) (ports.Dialer, error) { | |
| 98 | + | call := int(f.calls.Add(1)) - 1 | |
| 99 | + | f.mu.Lock() | |
| 100 | + | defer f.mu.Unlock() | |
| 101 | + | if call < 0 || call >= len(f.dialers) { | |
| 102 | + | return nil, io.EOF | |
| 103 | + | } | |
| 104 | + | return f.dialers[call], nil | |
| 105 | + | } | |
| 106 | + | ||
| 91 | 107 | func remoteLinkTestTarget() domain.RemoteSessionTarget { | |
| 92 | 108 | return domain.RemoteSessionTarget{ | |
| 93 | 109 | Endpoint: "remote.example", | |
| @@ -99,6 +115,11 @@ func remoteLinkTestTarget() domain.RemoteSessionTarget { | |||
| 99 | 115 | } | |
| 100 | 116 | ||
| 101 | 117 | func enqueueRemoteLinkHandshake(t *testing.T, transport *remoteLinkTestTransport, target domain.RemoteSessionTarget, size domain.Size) { | |
| 118 | + | t.Helper() | |
| 119 | + | enqueueRemoteLinkHandshakeWithContent(t, transport, target, size, "remote marker") | |
| 120 | + | } | |
| 121 | + | ||
| 122 | + | func enqueueRemoteLinkHandshakeWithContent(t *testing.T, transport *remoteLinkTestTransport, target domain.RemoteSessionTarget, size domain.Size, contentMarker string) { | |
| 102 | 123 | t.Helper() | |
| 103 | 124 | content := contentSize(size) | |
| 104 | 125 | metaPayload, err := ports.MarshalSessionMeta(ports.SessionMeta{ | |
| @@ -110,7 +131,7 @@ func enqueueRemoteLinkHandshake(t *testing.T, transport *remoteLinkTestTransport | |||
| 110 | 131 | }) | |
| 111 | 132 | require.NoError(t, err) | |
| 112 | 133 | outputPayload, err := ports.MarshalOutput(ports.Output{ | |
| 113 | - | Epoch: 1, New: 1, Size: content, Full: true, Data: []byte("remote marker"), | |
| 134 | + | Epoch: 1, New: 1, Size: content, Full: true, Data: []byte(contentMarker), | |
| 114 | 135 | }) | |
| 115 | 136 | require.NoError(t, err) | |
| 116 | 137 | transport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgWelcome, Payload: ports.MarshalWelcome(ports.Welcome{ | |
| @@ -338,6 +359,201 @@ func TestOpenRemoteViewPublishesHandshakeReadyCandidate(t *testing.T) { | |||
| 338 | 359 | require.NoError(t, transport.Close()) | |
| 339 | 360 | } | |
| 340 | 361 | ||
| 362 | + | func TestOpenRemoteViewReconnectsInactiveExactView(t *testing.T) { | |
| 363 | + | d := newTestDaemon(t, nil, stubClock{}) | |
| 364 | + | target := remoteLinkTestTarget() | |
| 365 | + | size := domain.Size{Cols: 80, Rows: 24} | |
| 366 | + | firstTransport := newRemoteLinkTestTransport() | |
| 367 | + | secondTransport := newRemoteLinkTestTransport() | |
| 368 | + | enqueueRemoteLinkHandshakeWithContent(t, firstTransport, target, size, "first generation") | |
| 369 | + | enqueueRemoteLinkHandshakeWithContent(t, secondTransport, target, size, "second generation") | |
| 370 | + | factory := &remoteLinkSequenceFactory{dialers: []ports.Dialer{ | |
| 371 | + | &remoteLinkTestDialer{transport: firstTransport}, | |
| 372 | + | &remoteLinkTestDialer{transport: secondTransport}, | |
| 373 | + | }} | |
| 374 | + | d.remoteDialerFactory = factory | |
| 375 | + | ||
| 376 | + | view, err := d.openRemoteView(context.Background(), target, size) | |
| 377 | + | require.NoError(t, err) | |
| 378 | + | awaitFrame(t, firstTransport.sent, ports.MsgHello) | |
| 379 | + | awaitFrame(t, firstTransport.sent, ports.MsgAck) | |
| 380 | + | ||
| 381 | + | localTransport := &closeTrackingTransport{} | |
| 382 | + | attachment := &attachedClient{tr: localTransport} | |
| 383 | + | attachment.setAttachmentOwner(view) | |
| 384 | + | view.mu.Lock() | |
| 385 | + | require.True(t, view.registerAttachmentLocked(attachment)) | |
| 386 | + | oldLink := view.link | |
| 387 | + | oldGeneration := view.linkGeneration | |
| 388 | + | oldLink.active = false | |
| 389 | + | require.Contains(t, screenLineText(view.screen, 0), "first generation") | |
| 390 | + | view.mu.Unlock() | |
| 391 | + | ||
| 392 | + | reconnected, err := d.openRemoteView(context.Background(), target, size) | |
| 393 | + | require.NoError(t, err) | |
| 394 | + | require.Same(t, view, reconnected) | |
| 395 | + | require.Equal(t, int32(2), factory.calls.Load(), "an inactive registry link must be re-established") | |
| 396 | + | awaitFrame(t, secondTransport.sent, ports.MsgHello) | |
| 397 | + | awaitFrame(t, secondTransport.sent, ports.MsgAck) | |
| 398 | + | ||
| 399 | + | view.mu.Lock() | |
| 400 | + | require.NotSame(t, oldLink, view.link) | |
| 401 | + | require.Equal(t, oldGeneration+1, view.linkGeneration) | |
| 402 | + | require.Same(t, view, view.link.view) | |
| 403 | + | require.Equal(t, view.linkGeneration, view.link.generation) | |
| 404 | + | require.True(t, view.link.active) | |
| 405 | + | require.Contains(t, screenLineText(view.screen, 0), "second generation") | |
| 406 | + | require.Contains(t, view.attachments, attachment, "reconnect must preserve local attachment membership") | |
| 407 | + | view.mu.Unlock() | |
| 408 | + | require.Same(t, view, attachment.currentAttachmentOwner()) | |
| 409 | + | select { | |
| 410 | + | case <-firstTransport.closed: | |
| 411 | + | default: | |
| 412 | + | t.Fatal("reconnect did not close the replaced exact transport") | |
| 413 | + | } | |
| 414 | + | awaitTestCompletion(t, oldLink.done, "reconnect did not join the replaced exact transport") | |
| 415 | + | ||
| 416 | + | d.shutdownAll(ports.ReasonServerShutdown) | |
| 417 | + | } | |
| 418 | + | ||
| 419 | + | func TestOpenRemoteViewFailedReconnectPreservesViewUntilValidatedPublication(t *testing.T) { | |
| 420 | + | d := newTestDaemon(t, nil, stubClock{}) | |
| 421 | + | target := remoteLinkTestTarget() | |
| 422 | + | size := domain.Size{Cols: 80, Rows: 24} | |
| 423 | + | firstTransport := newRemoteLinkTestTransport() | |
| 424 | + | failedTransport := newRemoteLinkTestTransport() | |
| 425 | + | enqueueRemoteLinkHandshakeWithContent(t, firstTransport, target, size, "retained generation") | |
| 426 | + | failedTransport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgWelcome, Payload: ports.MarshalWelcome(ports.Welcome{ | |
| 427 | + | SessionID: "remote-id", SessionName: target.SessionName, RenderMode: ports.RenderModeProxiedContent, | |
| 428 | + | })}} | |
| 429 | + | invalidMetadata, err := ports.MarshalSessionMeta(ports.SessionMeta{ | |
| 430 | + | LifecycleID: domain.SessionLifecycleID{9}, Revision: 1, SessionName: target.SessionName, ActiveTabID: target.LiveTabID, | |
| 431 | + | Tabs: []ports.SessionTabMeta{{ID: target.LiveTabID, Name: "main"}}, | |
| 432 | + | }) | |
| 433 | + | require.NoError(t, err) | |
| 434 | + | failedTransport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgSessionMeta, Payload: invalidMetadata}} | |
| 435 | + | factory := &remoteLinkSequenceFactory{dialers: []ports.Dialer{ | |
| 436 | + | &remoteLinkTestDialer{transport: firstTransport}, | |
| 437 | + | &remoteLinkTestDialer{transport: failedTransport}, | |
| 438 | + | }} | |
| 439 | + | d.remoteDialerFactory = factory | |
| 440 | + | ||
| 441 | + | view, err := d.openRemoteView(context.Background(), target, size) | |
| 442 | + | require.NoError(t, err) | |
| 443 | + | awaitFrame(t, firstTransport.sent, ports.MsgHello) | |
| 444 | + | awaitFrame(t, firstTransport.sent, ports.MsgAck) | |
| 445 | + | attachment := &attachedClient{tr: &closeTrackingTransport{}} | |
| 446 | + | attachment.setAttachmentOwner(view) | |
| 447 | + | view.mu.Lock() | |
| 448 | + | require.True(t, view.registerAttachmentLocked(attachment)) | |
| 449 | + | oldLink := view.link | |
| 450 | + | oldGeneration := view.linkGeneration | |
| 451 | + | oldScreen := view.screen | |
| 452 | + | oldLink.active = false | |
| 453 | + | view.mu.Unlock() | |
| 454 | + | ||
| 455 | + | reconnected, err := d.openRemoteView(context.Background(), target, size) | |
| 456 | + | require.Nil(t, reconnected) | |
| 457 | + | require.ErrorContains(t, err, "metadata identity mismatch") | |
| 458 | + | require.Equal(t, int32(2), factory.calls.Load(), "the inactive registry hit must not be returned") | |
| 459 | + | select { | |
| 460 | + | case <-failedTransport.closed: | |
| 461 | + | default: | |
| 462 | + | t.Fatal("failed reconnect candidate transport was not closed") | |
| 463 | + | } | |
| 464 | + | ||
| 465 | + | d.mu.Lock() | |
| 466 | + | require.Same(t, view, d.remoteViewByKeyLocked(view.key)) | |
| 467 | + | d.mu.Unlock() | |
| 468 | + | view.mu.Lock() | |
| 469 | + | require.Same(t, oldLink, view.link) | |
| 470 | + | require.Equal(t, oldGeneration, view.linkGeneration) | |
| 471 | + | require.Same(t, oldScreen, view.screen) | |
| 472 | + | require.Contains(t, screenLineText(view.screen, 0), "retained generation") | |
| 473 | + | require.Contains(t, view.attachments, attachment) | |
| 474 | + | view.mu.Unlock() | |
| 475 | + | require.Same(t, view, attachment.currentAttachmentOwner()) | |
| 476 | + | ||
| 477 | + | d.shutdownAll(ports.ReasonServerShutdown) | |
| 478 | + | } | |
| 479 | + | ||
| 480 | + | func TestOpenRemoteViewReconnectPublicationRejectsStaleLinkGeneration(t *testing.T) { | |
| 481 | + | d := newTestDaemon(t, nil, stubClock{}) | |
| 482 | + | target := remoteLinkTestTarget() | |
| 483 | + | size := domain.Size{Cols: 80, Rows: 24} | |
| 484 | + | oldTransport := newRemoteLinkTestTransport() | |
| 485 | + | candidateTransport := newRemoteLinkTestTransport() | |
| 486 | + | enqueueRemoteLinkHandshake(t, oldTransport, target, size) | |
| 487 | + | enqueueRemoteLinkHandshakeWithContent(t, candidateTransport, target, size, "stale candidate") | |
| 488 | + | reconnectStarted := make(chan struct{}) | |
| 489 | + | releaseReconnect := make(chan struct{}) | |
| 490 | + | factory := &remoteLinkSequenceFactory{dialers: []ports.Dialer{ | |
| 491 | + | &remoteLinkTestDialer{transport: oldTransport}, | |
| 492 | + | &remoteLinkTestDialer{transport: candidateTransport, started: reconnectStarted, release: releaseReconnect}, | |
| 493 | + | }} | |
| 494 | + | d.remoteDialerFactory = factory | |
| 495 | + | ||
| 496 | + | view, err := d.openRemoteView(context.Background(), target, size) | |
| 497 | + | require.NoError(t, err) | |
| 498 | + | awaitFrame(t, oldTransport.sent, ports.MsgHello) | |
| 499 | + | awaitFrame(t, oldTransport.sent, ports.MsgAck) | |
| 500 | + | view.mu.Lock() | |
| 501 | + | observedLink := view.link | |
| 502 | + | observedLink.active = false | |
| 503 | + | view.mu.Unlock() | |
| 504 | + | ||
| 505 | + | type result struct { | |
| 506 | + | view *remoteView | |
| 507 | + | err error | |
| 508 | + | } | |
| 509 | + | resultCh := make(chan result, 1) | |
| 510 | + | go func() { | |
| 511 | + | reconnected, reconnectErr := d.openRemoteView(context.Background(), target, size) | |
| 512 | + | resultCh <- result{view: reconnected, err: reconnectErr} | |
| 513 | + | }() | |
| 514 | + | awaitTestCompletion(t, reconnectStarted, "reconnect dial did not start") | |
| 515 | + | ||
| 516 | + | newerTransport := newRemoteLinkTestTransport() | |
| 517 | + | _, cancelNewer := context.WithCancel(context.Background()) | |
| 518 | + | view.mu.Lock() | |
| 519 | + | view.linkGeneration++ | |
| 520 | + | newerLink := &remoteLink{ | |
| 521 | + | view: view, generation: view.linkGeneration, target: target, transport: newerTransport, | |
| 522 | + | cancel: cancelNewer, done: make(chan struct{}), active: true, | |
| 523 | + | } | |
| 524 | + | view.link = newerLink | |
| 525 | + | view.mu.Unlock() | |
| 526 | + | d.startRemoteLink(newerLink) | |
| 527 | + | close(releaseReconnect) | |
| 528 | + | ||
| 529 | + | got := awaitTestValue(t, resultCh, "reconnect did not finish") | |
| 530 | + | require.NoError(t, got.err) | |
| 531 | + | require.Same(t, view, got.view) | |
| 532 | + | view.mu.Lock() | |
| 533 | + | require.Same(t, newerLink, view.link, "stale reconnect publication replaced a newer generation") | |
| 534 | + | view.mu.Unlock() | |
| 535 | + | select { | |
| 536 | + | case <-candidateTransport.closed: | |
| 537 | + | default: | |
| 538 | + | t.Fatal("stale reconnect candidate transport was not closed") | |
| 539 | + | } | |
| 540 | + | select { | |
| 541 | + | case <-newerTransport.closed: | |
| 542 | + | t.Fatal("stale reconnect closed the newer exact transport") | |
| 543 | + | default: | |
| 544 | + | } | |
| 545 | + | select { | |
| 546 | + | case <-oldTransport.closed: | |
| 547 | + | t.Fatal("rejected reconnect closed a transport it did not replace") | |
| 548 | + | default: | |
| 549 | + | } | |
| 550 | + | ||
| 551 | + | observedLink.cancel() | |
| 552 | + | require.NoError(t, oldTransport.Close()) | |
| 553 | + | awaitTestCompletion(t, observedLink.done, "superseded test link did not stop") | |
| 554 | + | d.shutdownAll(ports.ReasonServerShutdown) | |
| 555 | + | } | |
| 556 | + | ||
| 341 | 557 | func TestOpenRemoteViewCancellationClosesUnpublishedTransport(t *testing.T) { | |
| 342 | 558 | d := newTestDaemon(t, nil, stubClock{}) | |
| 343 | 559 | target := remoteLinkTestTarget() | |
| @@ -88,6 +88,22 @@ func (f *remoteLinkTestFactory) DialerForRemote(_ string, _ string, _ ports.Remo | |||||
| 88 | return f.dialer, nil | 88 | return f.dialer, nil | ||
| 89 | } | 89 | } | ||
| 90 | 90 | ||||
| 91 | + | type remoteLinkSequenceFactory struct { | |||
| 92 | + | mu sync.Mutex | |||
| 93 | + | dialers []ports.Dialer | |||
| 94 | + | calls atomic.Int32 | |||
| 95 | + | } | |||
| 96 | + | ||||
| 97 | + | func (f *remoteLinkSequenceFactory) DialerForRemote(_ string, _ string, _ ports.RemoteTransportMode, _ *slog.Logger) (ports.Dialer, error) { | |||
| 98 | + | call := int(f.calls.Add(1)) - 1 | |||
| 99 | + | f.mu.Lock() | |||
| 100 | + | defer f.mu.Unlock() | |||
| 101 | + | if call < 0 || call >= len(f.dialers) { | |||
| 102 | + | return nil, io.EOF | |||
| 103 | + | } | |||
| 104 | + | return f.dialers[call], nil | |||
| 105 | + | } | |||
| 106 | + | ||||
| 91 | func remoteLinkTestTarget() domain.RemoteSessionTarget { | 107 | func remoteLinkTestTarget() domain.RemoteSessionTarget { | ||
| 92 | return domain.RemoteSessionTarget{ | 108 | return domain.RemoteSessionTarget{ | ||
| 93 | Endpoint: "remote.example", | 109 | Endpoint: "remote.example", | ||
| @@ -99,6 +115,11 @@ func remoteLinkTestTarget() domain.RemoteSessionTarget { | |||||
| 99 | } | 115 | } | ||
| 100 | 116 | ||||
| 101 | func enqueueRemoteLinkHandshake(t *testing.T, transport *remoteLinkTestTransport, target domain.RemoteSessionTarget, size domain.Size) { | 117 | func enqueueRemoteLinkHandshake(t *testing.T, transport *remoteLinkTestTransport, target domain.RemoteSessionTarget, size domain.Size) { | ||
| 118 | + | t.Helper() | |||
| 119 | + | enqueueRemoteLinkHandshakeWithContent(t, transport, target, size, "remote marker") | |||
| 120 | + | } | |||
| 121 | + | ||||
| 122 | + | func enqueueRemoteLinkHandshakeWithContent(t *testing.T, transport *remoteLinkTestTransport, target domain.RemoteSessionTarget, size domain.Size, contentMarker string) { | |||
| 102 | t.Helper() | 123 | t.Helper() | ||
| 103 | content := contentSize(size) | 124 | content := contentSize(size) | ||
| 104 | metaPayload, err := ports.MarshalSessionMeta(ports.SessionMeta{ | 125 | metaPayload, err := ports.MarshalSessionMeta(ports.SessionMeta{ | ||
| @@ -110,7 +131,7 @@ func enqueueRemoteLinkHandshake(t *testing.T, transport *remoteLinkTestTransport | |||||
| 110 | }) | 131 | }) | ||
| 111 | require.NoError(t, err) | 132 | require.NoError(t, err) | ||
| 112 | outputPayload, err := ports.MarshalOutput(ports.Output{ | 133 | outputPayload, err := ports.MarshalOutput(ports.Output{ | ||
| 113 | - | Epoch: 1, New: 1, Size: content, Full: true, Data: []byte("remote marker"), | 134 | + | Epoch: 1, New: 1, Size: content, Full: true, Data: []byte(contentMarker), |
| 114 | }) | 135 | }) | ||
| 115 | require.NoError(t, err) | 136 | require.NoError(t, err) | ||
| 116 | transport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgWelcome, Payload: ports.MarshalWelcome(ports.Welcome{ | 137 | transport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgWelcome, Payload: ports.MarshalWelcome(ports.Welcome{ | ||
| @@ -338,6 +359,201 @@ func TestOpenRemoteViewPublishesHandshakeReadyCandidate(t *testing.T) { | |||||
| 338 | require.NoError(t, transport.Close()) | 359 | require.NoError(t, transport.Close()) | ||
| 339 | } | 360 | } | ||
| 340 | 361 | ||||
| 362 | + | func TestOpenRemoteViewReconnectsInactiveExactView(t *testing.T) { | |||
| 363 | + | d := newTestDaemon(t, nil, stubClock{}) | |||
| 364 | + | target := remoteLinkTestTarget() | |||
| 365 | + | size := domain.Size{Cols: 80, Rows: 24} | |||
| 366 | + | firstTransport := newRemoteLinkTestTransport() | |||
| 367 | + | secondTransport := newRemoteLinkTestTransport() | |||
| 368 | + | enqueueRemoteLinkHandshakeWithContent(t, firstTransport, target, size, "first generation") | |||
| 369 | + | enqueueRemoteLinkHandshakeWithContent(t, secondTransport, target, size, "second generation") | |||
| 370 | + | factory := &remoteLinkSequenceFactory{dialers: []ports.Dialer{ | |||
| 371 | + | &remoteLinkTestDialer{transport: firstTransport}, | |||
| 372 | + | &remoteLinkTestDialer{transport: secondTransport}, | |||
| 373 | + | }} | |||
| 374 | + | d.remoteDialerFactory = factory | |||
| 375 | + | ||||
| 376 | + | view, err := d.openRemoteView(context.Background(), target, size) | |||
| 377 | + | require.NoError(t, err) | |||
| 378 | + | awaitFrame(t, firstTransport.sent, ports.MsgHello) | |||
| 379 | + | awaitFrame(t, firstTransport.sent, ports.MsgAck) | |||
| 380 | + | ||||
| 381 | + | localTransport := &closeTrackingTransport{} | |||
| 382 | + | attachment := &attachedClient{tr: localTransport} | |||
| 383 | + | attachment.setAttachmentOwner(view) | |||
| 384 | + | view.mu.Lock() | |||
| 385 | + | require.True(t, view.registerAttachmentLocked(attachment)) | |||
| 386 | + | oldLink := view.link | |||
| 387 | + | oldGeneration := view.linkGeneration | |||
| 388 | + | oldLink.active = false | |||
| 389 | + | require.Contains(t, screenLineText(view.screen, 0), "first generation") | |||
| 390 | + | view.mu.Unlock() | |||
| 391 | + | ||||
| 392 | + | reconnected, err := d.openRemoteView(context.Background(), target, size) | |||
| 393 | + | require.NoError(t, err) | |||
| 394 | + | require.Same(t, view, reconnected) | |||
| 395 | + | require.Equal(t, int32(2), factory.calls.Load(), "an inactive registry link must be re-established") | |||
| 396 | + | awaitFrame(t, secondTransport.sent, ports.MsgHello) | |||
| 397 | + | awaitFrame(t, secondTransport.sent, ports.MsgAck) | |||
| 398 | + | ||||
| 399 | + | view.mu.Lock() | |||
| 400 | + | require.NotSame(t, oldLink, view.link) | |||
| 401 | + | require.Equal(t, oldGeneration+1, view.linkGeneration) | |||
| 402 | + | require.Same(t, view, view.link.view) | |||
| 403 | + | require.Equal(t, view.linkGeneration, view.link.generation) | |||
| 404 | + | require.True(t, view.link.active) | |||
| 405 | + | require.Contains(t, screenLineText(view.screen, 0), "second generation") | |||
| 406 | + | require.Contains(t, view.attachments, attachment, "reconnect must preserve local attachment membership") | |||
| 407 | + | view.mu.Unlock() | |||
| 408 | + | require.Same(t, view, attachment.currentAttachmentOwner()) | |||
| 409 | + | select { | |||
| 410 | + | case <-firstTransport.closed: | |||
| 411 | + | default: | |||
| 412 | + | t.Fatal("reconnect did not close the replaced exact transport") | |||
| 413 | + | } | |||
| 414 | + | awaitTestCompletion(t, oldLink.done, "reconnect did not join the replaced exact transport") | |||
| 415 | + | ||||
| 416 | + | d.shutdownAll(ports.ReasonServerShutdown) | |||
| 417 | + | } | |||
| 418 | + | ||||
| 419 | + | func TestOpenRemoteViewFailedReconnectPreservesViewUntilValidatedPublication(t *testing.T) { | |||
| 420 | + | d := newTestDaemon(t, nil, stubClock{}) | |||
| 421 | + | target := remoteLinkTestTarget() | |||
| 422 | + | size := domain.Size{Cols: 80, Rows: 24} | |||
| 423 | + | firstTransport := newRemoteLinkTestTransport() | |||
| 424 | + | failedTransport := newRemoteLinkTestTransport() | |||
| 425 | + | enqueueRemoteLinkHandshakeWithContent(t, firstTransport, target, size, "retained generation") | |||
| 426 | + | failedTransport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgWelcome, Payload: ports.MarshalWelcome(ports.Welcome{ | |||
| 427 | + | SessionID: "remote-id", SessionName: target.SessionName, RenderMode: ports.RenderModeProxiedContent, | |||
| 428 | + | })}} | |||
| 429 | + | invalidMetadata, err := ports.MarshalSessionMeta(ports.SessionMeta{ | |||
| 430 | + | LifecycleID: domain.SessionLifecycleID{9}, Revision: 1, SessionName: target.SessionName, ActiveTabID: target.LiveTabID, | |||
| 431 | + | Tabs: []ports.SessionTabMeta{{ID: target.LiveTabID, Name: "main"}}, | |||
| 432 | + | }) | |||
| 433 | + | require.NoError(t, err) | |||
| 434 | + | failedTransport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgSessionMeta, Payload: invalidMetadata}} | |||
| 435 | + | factory := &remoteLinkSequenceFactory{dialers: []ports.Dialer{ | |||
| 436 | + | &remoteLinkTestDialer{transport: firstTransport}, | |||
| 437 | + | &remoteLinkTestDialer{transport: failedTransport}, | |||
| 438 | + | }} | |||
| 439 | + | d.remoteDialerFactory = factory | |||
| 440 | + | ||||
| 441 | + | view, err := d.openRemoteView(context.Background(), target, size) | |||
| 442 | + | require.NoError(t, err) | |||
| 443 | + | awaitFrame(t, firstTransport.sent, ports.MsgHello) | |||
| 444 | + | awaitFrame(t, firstTransport.sent, ports.MsgAck) | |||
| 445 | + | attachment := &attachedClient{tr: &closeTrackingTransport{}} | |||
| 446 | + | attachment.setAttachmentOwner(view) | |||
| 447 | + | view.mu.Lock() | |||
| 448 | + | require.True(t, view.registerAttachmentLocked(attachment)) | |||
| 449 | + | oldLink := view.link | |||
| 450 | + | oldGeneration := view.linkGeneration | |||
| 451 | + | oldScreen := view.screen | |||
| 452 | + | oldLink.active = false | |||
| 453 | + | view.mu.Unlock() | |||
| 454 | + | ||||
| 455 | + | reconnected, err := d.openRemoteView(context.Background(), target, size) | |||
| 456 | + | require.Nil(t, reconnected) | |||
| 457 | + | require.ErrorContains(t, err, "metadata identity mismatch") | |||
| 458 | + | require.Equal(t, int32(2), factory.calls.Load(), "the inactive registry hit must not be returned") | |||
| 459 | + | select { | |||
| 460 | + | case <-failedTransport.closed: | |||
| 461 | + | default: | |||
| 462 | + | t.Fatal("failed reconnect candidate transport was not closed") | |||
| 463 | + | } | |||
| 464 | + | ||||
| 465 | + | d.mu.Lock() | |||
| 466 | + | require.Same(t, view, d.remoteViewByKeyLocked(view.key)) | |||
| 467 | + | d.mu.Unlock() | |||
| 468 | + | view.mu.Lock() | |||
| 469 | + | require.Same(t, oldLink, view.link) | |||
| 470 | + | require.Equal(t, oldGeneration, view.linkGeneration) | |||
| 471 | + | require.Same(t, oldScreen, view.screen) | |||
| 472 | + | require.Contains(t, screenLineText(view.screen, 0), "retained generation") | |||
| 473 | + | require.Contains(t, view.attachments, attachment) | |||
| 474 | + | view.mu.Unlock() | |||
| 475 | + | require.Same(t, view, attachment.currentAttachmentOwner()) | |||
| 476 | + | ||||
| 477 | + | d.shutdownAll(ports.ReasonServerShutdown) | |||
| 478 | + | } | |||
| 479 | + | ||||
| 480 | + | func TestOpenRemoteViewReconnectPublicationRejectsStaleLinkGeneration(t *testing.T) { | |||
| 481 | + | d := newTestDaemon(t, nil, stubClock{}) | |||
| 482 | + | target := remoteLinkTestTarget() | |||
| 483 | + | size := domain.Size{Cols: 80, Rows: 24} | |||
| 484 | + | oldTransport := newRemoteLinkTestTransport() | |||
| 485 | + | candidateTransport := newRemoteLinkTestTransport() | |||
| 486 | + | enqueueRemoteLinkHandshake(t, oldTransport, target, size) | |||
| 487 | + | enqueueRemoteLinkHandshakeWithContent(t, candidateTransport, target, size, "stale candidate") | |||
| 488 | + | reconnectStarted := make(chan struct{}) | |||
| 489 | + | releaseReconnect := make(chan struct{}) | |||
| 490 | + | factory := &remoteLinkSequenceFactory{dialers: []ports.Dialer{ | |||
| 491 | + | &remoteLinkTestDialer{transport: oldTransport}, | |||
| 492 | + | &remoteLinkTestDialer{transport: candidateTransport, started: reconnectStarted, release: releaseReconnect}, | |||
| 493 | + | }} | |||
| 494 | + | d.remoteDialerFactory = factory | |||
| 495 | + | ||||
| 496 | + | view, err := d.openRemoteView(context.Background(), target, size) | |||
| 497 | + | require.NoError(t, err) | |||
| 498 | + | awaitFrame(t, oldTransport.sent, ports.MsgHello) | |||
| 499 | + | awaitFrame(t, oldTransport.sent, ports.MsgAck) | |||
| 500 | + | view.mu.Lock() | |||
| 501 | + | observedLink := view.link | |||
| 502 | + | observedLink.active = false | |||
| 503 | + | view.mu.Unlock() | |||
| 504 | + | ||||
| 505 | + | type result struct { | |||
| 506 | + | view *remoteView | |||
| 507 | + | err error | |||
| 508 | + | } | |||
| 509 | + | resultCh := make(chan result, 1) | |||
| 510 | + | go func() { | |||
| 511 | + | reconnected, reconnectErr := d.openRemoteView(context.Background(), target, size) | |||
| 512 | + | resultCh <- result{view: reconnected, err: reconnectErr} | |||
| 513 | + | }() | |||
| 514 | + | awaitTestCompletion(t, reconnectStarted, "reconnect dial did not start") | |||
| 515 | + | ||||
| 516 | + | newerTransport := newRemoteLinkTestTransport() | |||
| 517 | + | _, cancelNewer := context.WithCancel(context.Background()) | |||
| 518 | + | view.mu.Lock() | |||
| 519 | + | view.linkGeneration++ | |||
| 520 | + | newerLink := &remoteLink{ | |||
| 521 | + | view: view, generation: view.linkGeneration, target: target, transport: newerTransport, | |||
| 522 | + | cancel: cancelNewer, done: make(chan struct{}), active: true, | |||
| 523 | + | } | |||
| 524 | + | view.link = newerLink | |||
| 525 | + | view.mu.Unlock() | |||
| 526 | + | d.startRemoteLink(newerLink) | |||
| 527 | + | close(releaseReconnect) | |||
| 528 | + | ||||
| 529 | + | got := awaitTestValue(t, resultCh, "reconnect did not finish") | |||
| 530 | + | require.NoError(t, got.err) | |||
| 531 | + | require.Same(t, view, got.view) | |||
| 532 | + | view.mu.Lock() | |||
| 533 | + | require.Same(t, newerLink, view.link, "stale reconnect publication replaced a newer generation") | |||
| 534 | + | view.mu.Unlock() | |||
| 535 | + | select { | |||
| 536 | + | case <-candidateTransport.closed: | |||
| 537 | + | default: | |||
| 538 | + | t.Fatal("stale reconnect candidate transport was not closed") | |||
| 539 | + | } | |||
| 540 | + | select { | |||
| 541 | + | case <-newerTransport.closed: | |||
| 542 | + | t.Fatal("stale reconnect closed the newer exact transport") | |||
| 543 | + | default: | |||
| 544 | + | } | |||
| 545 | + | select { | |||
| 546 | + | case <-oldTransport.closed: | |||
| 547 | + | t.Fatal("rejected reconnect closed a transport it did not replace") | |||
| 548 | + | default: | |||
| 549 | + | } | |||
| 550 | + | ||||
| 551 | + | observedLink.cancel() | |||
| 552 | + | require.NoError(t, oldTransport.Close()) | |||
| 553 | + | awaitTestCompletion(t, observedLink.done, "superseded test link did not stop") | |||
| 554 | + | d.shutdownAll(ports.ReasonServerShutdown) | |||
| 555 | + | } | |||
| 556 | + | ||||
| 341 | func TestOpenRemoteViewCancellationClosesUnpublishedTransport(t *testing.T) { | 557 | func TestOpenRemoteViewCancellationClosesUnpublishedTransport(t *testing.T) { | ||
| 342 | d := newTestDaemon(t, nil, stubClock{}) | 558 | d := newTestDaemon(t, nil, stubClock{}) | ||
| 343 | target := remoteLinkTestTarget() | 559 | target := remoteLinkTestTarget() | ||