fix(daemon): recover remote resume before parking
- commit
- fe43a19128de4a2f742fe466926be6e12757ada3
- author
- brice · · 2026-07-29 20:09 UTC
- parent
- 8eb2673
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
5 files · +422 -11
M internal/usecase/daemon/client_transport_errors.go +19 -8
| @@ -5,11 +5,18 @@ import "github.com/bnema/vev/internal/ports" | |||
| 5 | 5 | // clientGone detaches ac if it is still the session's current client. The | |
| 6 | 6 | // session remains registered and headless after the client is gone. | |
| 7 | 7 | func (d *Daemon) clientGone(sess *session, ac *attachedClient, failed ports.Transport, explicit bool) { | |
| 8 | - | if failed != nil && !ac.currentTransportIs(failed) { | |
| 9 | - | return // stale connection loop; a newer transport owns this client | |
| 8 | + | expected := transportSnapshot{} | |
| 9 | + | if failed != nil { | |
| 10 | + | expected = ac.transportSnapshot() | |
| 11 | + | if expected.transport != failed { | |
| 12 | + | return // stale connection loop; a newer transport owns this client | |
| 13 | + | } | |
| 10 | 14 | } | |
| 11 | - | if !d.detachIfCurrent(sess, ac) { | |
| 12 | - | return // already displaced by a newer client; nothing to do | |
| 15 | + | if d.beforeClientGoneDetach != nil { | |
| 16 | + | d.beforeClientGoneDetach() | |
| 17 | + | } | |
| 18 | + | if !d.detachIfCurrentTransport(sess, ac, expected) { | |
| 19 | + | return // displaced, or the link was rebound after the precheck | |
| 13 | 20 | } | |
| 14 | 21 | d.finishClientGone(sess, ac, failed, explicit) | |
| 15 | 22 | } | |
| @@ -65,10 +72,14 @@ func (d *Daemon) finishClientGone(sess *session, ac *attachedClient, failed port | |||
| 65 | 72 | // detachOnSendError drops a client whose transport failed, leaving the session | |
| 66 | 73 | // registered and headless. | |
| 67 | 74 | func (d *Daemon) detachOnSendError(sess *session, ac *attachedClient, failed ports.Transport) { | |
| 68 | - | if failed != nil && !ac.currentTransportIs(failed) { | |
| 69 | - | return | |
| 70 | - | } | |
| 71 | - | if d.detachIfCurrent(sess, ac) { | |
| 75 | + | expected := transportSnapshot{} | |
| 76 | + | if failed != nil { | |
| 77 | + | expected = ac.transportSnapshot() | |
| 78 | + | if expected.transport != failed { | |
| 79 | + | return | |
| 80 | + | } | |
| 81 | + | } | |
| 82 | + | if d.detachIfCurrentTransport(sess, ac, expected) { | |
| 72 | 83 | d.finishSendErrorDetach(sess, ac, failed) | |
| 73 | 84 | } | |
| 74 | 85 | } | |
| @@ -5,11 +5,18 @@ import "github.com/bnema/vev/internal/ports" | |||||
| 5 | // clientGone detaches ac if it is still the session's current client. The | 5 | // clientGone detaches ac if it is still the session's current client. The | ||
| 6 | // session remains registered and headless after the client is gone. | 6 | // session remains registered and headless after the client is gone. | ||
| 7 | func (d *Daemon) clientGone(sess *session, ac *attachedClient, failed ports.Transport, explicit bool) { | 7 | func (d *Daemon) clientGone(sess *session, ac *attachedClient, failed ports.Transport, explicit bool) { | ||
| 8 | - | if failed != nil && !ac.currentTransportIs(failed) { | 8 | + | expected := transportSnapshot{} |
| 9 | - | return // stale connection loop; a newer transport owns this client | 9 | + | if failed != nil { |
| 10 | + | expected = ac.transportSnapshot() | |||
| 11 | + | if expected.transport != failed { | |||
| 12 | + | return // stale connection loop; a newer transport owns this client | |||
| 13 | + | } | |||
| 10 | } | 14 | } | ||
| 11 | - | if !d.detachIfCurrent(sess, ac) { | 15 | + | if d.beforeClientGoneDetach != nil { |
| 12 | - | return // already displaced by a newer client; nothing to do | 16 | + | d.beforeClientGoneDetach() |
| 17 | + | } | |||
| 18 | + | if !d.detachIfCurrentTransport(sess, ac, expected) { | |||
| 19 | + | return // displaced, or the link was rebound after the precheck | |||
| 13 | } | 20 | } | ||
| 14 | d.finishClientGone(sess, ac, failed, explicit) | 21 | d.finishClientGone(sess, ac, failed, explicit) | ||
| 15 | } | 22 | } | ||
| @@ -65,10 +72,14 @@ func (d *Daemon) finishClientGone(sess *session, ac *attachedClient, failed port | |||||
| 65 | // detachOnSendError drops a client whose transport failed, leaving the session | 72 | // detachOnSendError drops a client whose transport failed, leaving the session | ||
| 66 | // registered and headless. | 73 | // registered and headless. | ||
| 67 | func (d *Daemon) detachOnSendError(sess *session, ac *attachedClient, failed ports.Transport) { | 74 | func (d *Daemon) detachOnSendError(sess *session, ac *attachedClient, failed ports.Transport) { | ||
| 68 | - | if failed != nil && !ac.currentTransportIs(failed) { | 75 | + | expected := transportSnapshot{} |
| 69 | - | return | 76 | + | if failed != nil { |
| 70 | - | } | 77 | + | expected = ac.transportSnapshot() |
| 71 | - | if d.detachIfCurrent(sess, ac) { | 78 | + | if expected.transport != failed { |
| 79 | + | return | |||
| 80 | + | } | |||
| 81 | + | } | |||
| 82 | + | if d.detachIfCurrentTransport(sess, ac, expected) { | |||
| 72 | d.finishSendErrorDetach(sess, ac, failed) | 83 | d.finishSendErrorDetach(sess, ac, failed) | ||
| 73 | } | 84 | } | ||
| 74 | } | 85 | } | ||
M internal/usecase/daemon/daemon.go +23 -3
| @@ -172,6 +172,13 @@ type Daemon struct { | |||
| 172 | 172 | // afterDetachRoleEffectsFrozen observes terminal detach after it wins the | |
| 173 | 173 | // attachment gate but before it checks session ownership. | |
| 174 | 174 | afterDetachRoleEffectsFrozen func() | |
| 175 | + | // beforeClientGoneDetach pauses clientGone after the stale-transport | |
| 176 | + | // precheck and before exact transport/incarnation detach validation. | |
| 177 | + | beforeClientGoneDetach func() | |
| 178 | + | // beforeResumeParkedSendMu pauses resumeParked after the initial parked | |
| 179 | + | // lookup and before the attachment send lock, so tests can consume or | |
| 180 | + | // replace the credential while the handshake waits. | |
| 181 | + | beforeResumeParkedSendMu func() | |
| 175 | 182 | // afterDisplacedCleanupStarted observes deferred displaced cleanup before it | |
| 176 | 183 | // synchronizes with the attachment gate. | |
| 177 | 184 | afterDisplacedCleanupStarted func() | |
| @@ -1171,11 +1178,24 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||
| 1171 | 1178 | // A non-zero token is an authoritative resume credential. If it is unknown, | |
| 1172 | 1179 | // expired, or raced with lifecycle teardown, fail closed instead of routing | |
| 1173 | 1180 | // the Hello as an ordinary attach that could create or replace ownership. | |
| 1181 | + | // The one legitimate pre-park race is a still-active same-client credential | |
| 1182 | + | // for the requested session; hand that into the park/resume lifecycle. | |
| 1174 | 1183 | if h.ResumeToken != 0 { | |
| 1175 | 1184 | d.mu.Lock() | |
| 1176 | 1185 | parkedAtStart := d.parked[h.ResumeToken] | |
| 1177 | 1186 | d.mu.Unlock() | |
| 1178 | 1187 | if parkedAtStart == nil { | |
| 1188 | + | if sess, ac, ok, err := d.resumeLiveAttachment(h, tr, sz); err != nil { | |
| 1189 | + | if errors.Is(err, errResumeTokenLifecycleRace) { | |
| 1190 | + | // Live recovery parked then lost a competing resumeParked | |
| 1191 | + | // race; keep the fail-closed wire response instead of | |
| 1192 | + | // leaking the internal sentinel to handleHello as ErrInternal. | |
| 1193 | + | return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | |
| 1194 | + | } | |
| 1195 | + | return nil, nil, err | |
| 1196 | + | } else if ok { | |
| 1197 | + | return sess, ac, nil | |
| 1198 | + | } | |
| 1179 | 1199 | return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | |
| 1180 | 1200 | } | |
| 1181 | 1201 | if sess, ac, ok, err := d.resumeParked(h, tr, sz); err == nil { | |
| @@ -1185,9 +1205,9 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||
| 1185 | 1205 | return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | |
| 1186 | 1206 | } else if errors.Is(err, errResumeTokenLifecycleRace) { | |
| 1187 | 1207 | // The parked entry was replaced while this handshake waited for | |
| 1188 | - | // its send lock. The ordinary attach path can reclaim the still-live | |
| 1189 | - | // named session instead of turning this grace-period race into a | |
| 1190 | - | // terminal no-such-session error. | |
| 1208 | + | // its send lock. Fail closed on the original credential rather | |
| 1209 | + | // than falling through to ordinary attach/create routing. | |
| 1210 | + | return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | |
| 1191 | 1211 | } else { | |
| 1192 | 1212 | return nil, nil, err | |
| 1193 | 1213 | } | |
| @@ -172,6 +172,13 @@ type Daemon struct { | |||||
| 172 | // afterDetachRoleEffectsFrozen observes terminal detach after it wins the | 172 | // afterDetachRoleEffectsFrozen observes terminal detach after it wins the | ||
| 173 | // attachment gate but before it checks session ownership. | 173 | // attachment gate but before it checks session ownership. | ||
| 174 | afterDetachRoleEffectsFrozen func() | 174 | afterDetachRoleEffectsFrozen func() | ||
| 175 | + | // beforeClientGoneDetach pauses clientGone after the stale-transport | |||
| 176 | + | // precheck and before exact transport/incarnation detach validation. | |||
| 177 | + | beforeClientGoneDetach func() | |||
| 178 | + | // beforeResumeParkedSendMu pauses resumeParked after the initial parked | |||
| 179 | + | // lookup and before the attachment send lock, so tests can consume or | |||
| 180 | + | // replace the credential while the handshake waits. | |||
| 181 | + | beforeResumeParkedSendMu func() | |||
| 175 | // afterDisplacedCleanupStarted observes deferred displaced cleanup before it | 182 | // afterDisplacedCleanupStarted observes deferred displaced cleanup before it | ||
| 176 | // synchronizes with the attachment gate. | 183 | // synchronizes with the attachment gate. | ||
| 177 | afterDisplacedCleanupStarted func() | 184 | afterDisplacedCleanupStarted func() | ||
| @@ -1171,11 +1178,24 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||||
| 1171 | // A non-zero token is an authoritative resume credential. If it is unknown, | 1178 | // A non-zero token is an authoritative resume credential. If it is unknown, | ||
| 1172 | // expired, or raced with lifecycle teardown, fail closed instead of routing | 1179 | // expired, or raced with lifecycle teardown, fail closed instead of routing | ||
| 1173 | // the Hello as an ordinary attach that could create or replace ownership. | 1180 | // the Hello as an ordinary attach that could create or replace ownership. | ||
| 1181 | + | // The one legitimate pre-park race is a still-active same-client credential | |||
| 1182 | + | // for the requested session; hand that into the park/resume lifecycle. | |||
| 1174 | if h.ResumeToken != 0 { | 1183 | if h.ResumeToken != 0 { | ||
| 1175 | d.mu.Lock() | 1184 | d.mu.Lock() | ||
| 1176 | parkedAtStart := d.parked[h.ResumeToken] | 1185 | parkedAtStart := d.parked[h.ResumeToken] | ||
| 1177 | d.mu.Unlock() | 1186 | d.mu.Unlock() | ||
| 1178 | if parkedAtStart == nil { | 1187 | if parkedAtStart == nil { | ||
| 1188 | + | if sess, ac, ok, err := d.resumeLiveAttachment(h, tr, sz); err != nil { | |||
| 1189 | + | if errors.Is(err, errResumeTokenLifecycleRace) { | |||
| 1190 | + | // Live recovery parked then lost a competing resumeParked | |||
| 1191 | + | // race; keep the fail-closed wire response instead of | |||
| 1192 | + | // leaking the internal sentinel to handleHello as ErrInternal. | |||
| 1193 | + | return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | |||
| 1194 | + | } | |||
| 1195 | + | return nil, nil, err | |||
| 1196 | + | } else if ok { | |||
| 1197 | + | return sess, ac, nil | |||
| 1198 | + | } | |||
| 1179 | return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | 1199 | return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | ||
| 1180 | } | 1200 | } | ||
| 1181 | if sess, ac, ok, err := d.resumeParked(h, tr, sz); err == nil { | 1201 | if sess, ac, ok, err := d.resumeParked(h, tr, sz); err == nil { | ||
| @@ -1185,9 +1205,9 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||||
| 1185 | return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | 1205 | return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | ||
| 1186 | } else if errors.Is(err, errResumeTokenLifecycleRace) { | 1206 | } else if errors.Is(err, errResumeTokenLifecycleRace) { | ||
| 1187 | // The parked entry was replaced while this handshake waited for | 1207 | // The parked entry was replaced while this handshake waited for | ||
| 1188 | - | // its send lock. The ordinary attach path can reclaim the still-live | 1208 | + | // its send lock. Fail closed on the original credential rather |
| 1189 | - | // named session instead of turning this grace-period race into a | 1209 | + | // than falling through to ordinary attach/create routing. |
| 1190 | - | // terminal no-such-session error. | 1210 | + | return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} |
| 1191 | } else { | 1211 | } else { | ||
| 1192 | return nil, nil, err | 1212 | return nil, nil, err | ||
| 1193 | } | 1213 | } | ||
M internal/usecase/daemon/resume.go +83 -0
| @@ -205,6 +205,86 @@ func finishParkedAttachmentRetirements(retirements []parkedAttachmentRetirement) | |||
| 205 | 205 | } | |
| 206 | 206 | } | |
| 207 | 207 | ||
| 208 | + | // resumeLiveAttachment recovers a nonzero resume credential that still belongs | |
| 209 | + | // to the named session's active attachment because transport teardown has not | |
| 210 | + | // parked it yet. Only the exact live owner (same session, token, and client ID) | |
| 211 | + | // is accepted; arbitrary unknown tokens stay fail-closed. | |
| 212 | + | func (d *Daemon) resumeLiveAttachment(h ports.Hello, tr ports.Transport, sz domain.Size) (*session, *attachedClient, bool, error) { | |
| 213 | + | d.mu.Lock() | |
| 214 | + | if d.closing { | |
| 215 | + | d.mu.Unlock() | |
| 216 | + | return nil, nil, false, &protoErr{ports.ErrServerShutdown, "daemon is shutting down"} | |
| 217 | + | } | |
| 218 | + | // Teardown may have parked the credential already. | |
| 219 | + | if d.parked[h.ResumeToken] != nil { | |
| 220 | + | d.mu.Unlock() | |
| 221 | + | return d.resumeParked(h, tr, sz) | |
| 222 | + | } | |
| 223 | + | sess := d.findByNameLocked(h.Name) | |
| 224 | + | var ( | |
| 225 | + | ac *attachedClient | |
| 226 | + | oldSnap transportSnapshot | |
| 227 | + | credentialMatch bool | |
| 228 | + | clientMismatch bool | |
| 229 | + | ) | |
| 230 | + | if sess != nil { | |
| 231 | + | sess.mu.Lock() | |
| 232 | + | ac = sess.client | |
| 233 | + | // resumeToken writes run under d.mu; read the live credential here | |
| 234 | + | // before releasing the daemon lock, then revalidate via the exact | |
| 235 | + | // transport incarnation through detachIfCurrentTransport. | |
| 236 | + | if ac != nil && ac.resumeCapable && ac.resumeToken == h.ResumeToken { | |
| 237 | + | if ac.clientID != h.ClientID { | |
| 238 | + | clientMismatch = true | |
| 239 | + | } else { | |
| 240 | + | oldSnap = ac.transportSnapshot() | |
| 241 | + | credentialMatch = oldSnap.transport != nil | |
| 242 | + | } | |
| 243 | + | } | |
| 244 | + | sess.mu.Unlock() | |
| 245 | + | } | |
| 246 | + | // Teardown may have parked the credential while we resolved the session. | |
| 247 | + | if d.parked[h.ResumeToken] != nil { | |
| 248 | + | d.mu.Unlock() | |
| 249 | + | return d.resumeParked(h, tr, sz) | |
| 250 | + | } | |
| 251 | + | d.mu.Unlock() | |
| 252 | + | ||
| 253 | + | if clientMismatch { | |
| 254 | + | d.log.Warn("resume rejected", "session", sess.name, "err", "client id mismatch") | |
| 255 | + | return nil, nil, false, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | |
| 256 | + | } | |
| 257 | + | if !credentialMatch { | |
| 258 | + | return nil, nil, false, nil | |
| 259 | + | } | |
| 260 | + | ||
| 261 | + | if !d.detachIfCurrentTransport(sess, ac, oldSnap) { | |
| 262 | + | // A newer owner, rebound link, or teardown won; try the parked path. | |
| 263 | + | return d.resumeParked(h, tr, sz) | |
| 264 | + | } | |
| 265 | + | if rc := sess.renderCoordinator(); rc != nil { | |
| 266 | + | rc.noteDetach(ac) | |
| 267 | + | } | |
| 268 | + | d.unregisterPreview(ac) | |
| 269 | + | if !d.parkAttachment(sess, ac) { | |
| 270 | + | // Detach already published; retire the captured link exactly once so | |
| 271 | + | // shutdown cannot leave an orphaned open transport with no owner. | |
| 272 | + | d.resetScreenDefaultColors(sess) | |
| 273 | + | ac.clearPreviousSession() | |
| 274 | + | _ = ac.closeCapturedTransport(ac.revokeTransport(oldSnap.transport)) | |
| 275 | + | d.mu.Lock() | |
| 276 | + | closing := d.closing | |
| 277 | + | d.mu.Unlock() | |
| 278 | + | if closing { | |
| 279 | + | return nil, nil, false, &protoErr{ports.ErrServerShutdown, "daemon is shutting down"} | |
| 280 | + | } | |
| 281 | + | return nil, nil, false, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | |
| 282 | + | } | |
| 283 | + | _ = ac.closeCapturedTransport(oldSnap.transport) | |
| 284 | + | d.log.Info("live resume credential parked for reconnect", "session", sess.name) | |
| 285 | + | return d.resumeParked(h, tr, sz) | |
| 286 | + | } | |
| 287 | + | ||
| 208 | 288 | // resumeParked acquires the parked client's output lock before the daemon | |
| 209 | 289 | // registry lock, preserving the global sendMu > Daemon.mu ordering. The parked | |
| 210 | 290 | // entry is revalidated after both locks are held because it may expire between | |
| @@ -218,6 +298,9 @@ func (d *Daemon) resumeParked(h ports.Hello, tr ports.Transport, sz domain.Size) | |||
| 218 | 298 | } | |
| 219 | 299 | ||
| 220 | 300 | ac := parked.ac | |
| 301 | + | if d.beforeResumeParkedSendMu != nil { | |
| 302 | + | d.beforeResumeParkedSendMu() | |
| 303 | + | } | |
| 221 | 304 | ac.sendMu.Lock() | |
| 222 | 305 | defer ac.sendMu.Unlock() | |
| 223 | 306 | d.mu.Lock() | |
| @@ -205,6 +205,86 @@ func finishParkedAttachmentRetirements(retirements []parkedAttachmentRetirement) | |||||
| 205 | } | 205 | } | ||
| 206 | } | 206 | } | ||
| 207 | 207 | ||||
| 208 | + | // resumeLiveAttachment recovers a nonzero resume credential that still belongs | |||
| 209 | + | // to the named session's active attachment because transport teardown has not | |||
| 210 | + | // parked it yet. Only the exact live owner (same session, token, and client ID) | |||
| 211 | + | // is accepted; arbitrary unknown tokens stay fail-closed. | |||
| 212 | + | func (d *Daemon) resumeLiveAttachment(h ports.Hello, tr ports.Transport, sz domain.Size) (*session, *attachedClient, bool, error) { | |||
| 213 | + | d.mu.Lock() | |||
| 214 | + | if d.closing { | |||
| 215 | + | d.mu.Unlock() | |||
| 216 | + | return nil, nil, false, &protoErr{ports.ErrServerShutdown, "daemon is shutting down"} | |||
| 217 | + | } | |||
| 218 | + | // Teardown may have parked the credential already. | |||
| 219 | + | if d.parked[h.ResumeToken] != nil { | |||
| 220 | + | d.mu.Unlock() | |||
| 221 | + | return d.resumeParked(h, tr, sz) | |||
| 222 | + | } | |||
| 223 | + | sess := d.findByNameLocked(h.Name) | |||
| 224 | + | var ( | |||
| 225 | + | ac *attachedClient | |||
| 226 | + | oldSnap transportSnapshot | |||
| 227 | + | credentialMatch bool | |||
| 228 | + | clientMismatch bool | |||
| 229 | + | ) | |||
| 230 | + | if sess != nil { | |||
| 231 | + | sess.mu.Lock() | |||
| 232 | + | ac = sess.client | |||
| 233 | + | // resumeToken writes run under d.mu; read the live credential here | |||
| 234 | + | // before releasing the daemon lock, then revalidate via the exact | |||
| 235 | + | // transport incarnation through detachIfCurrentTransport. | |||
| 236 | + | if ac != nil && ac.resumeCapable && ac.resumeToken == h.ResumeToken { | |||
| 237 | + | if ac.clientID != h.ClientID { | |||
| 238 | + | clientMismatch = true | |||
| 239 | + | } else { | |||
| 240 | + | oldSnap = ac.transportSnapshot() | |||
| 241 | + | credentialMatch = oldSnap.transport != nil | |||
| 242 | + | } | |||
| 243 | + | } | |||
| 244 | + | sess.mu.Unlock() | |||
| 245 | + | } | |||
| 246 | + | // Teardown may have parked the credential while we resolved the session. | |||
| 247 | + | if d.parked[h.ResumeToken] != nil { | |||
| 248 | + | d.mu.Unlock() | |||
| 249 | + | return d.resumeParked(h, tr, sz) | |||
| 250 | + | } | |||
| 251 | + | d.mu.Unlock() | |||
| 252 | + | ||||
| 253 | + | if clientMismatch { | |||
| 254 | + | d.log.Warn("resume rejected", "session", sess.name, "err", "client id mismatch") | |||
| 255 | + | return nil, nil, false, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | |||
| 256 | + | } | |||
| 257 | + | if !credentialMatch { | |||
| 258 | + | return nil, nil, false, nil | |||
| 259 | + | } | |||
| 260 | + | ||||
| 261 | + | if !d.detachIfCurrentTransport(sess, ac, oldSnap) { | |||
| 262 | + | // A newer owner, rebound link, or teardown won; try the parked path. | |||
| 263 | + | return d.resumeParked(h, tr, sz) | |||
| 264 | + | } | |||
| 265 | + | if rc := sess.renderCoordinator(); rc != nil { | |||
| 266 | + | rc.noteDetach(ac) | |||
| 267 | + | } | |||
| 268 | + | d.unregisterPreview(ac) | |||
| 269 | + | if !d.parkAttachment(sess, ac) { | |||
| 270 | + | // Detach already published; retire the captured link exactly once so | |||
| 271 | + | // shutdown cannot leave an orphaned open transport with no owner. | |||
| 272 | + | d.resetScreenDefaultColors(sess) | |||
| 273 | + | ac.clearPreviousSession() | |||
| 274 | + | _ = ac.closeCapturedTransport(ac.revokeTransport(oldSnap.transport)) | |||
| 275 | + | d.mu.Lock() | |||
| 276 | + | closing := d.closing | |||
| 277 | + | d.mu.Unlock() | |||
| 278 | + | if closing { | |||
| 279 | + | return nil, nil, false, &protoErr{ports.ErrServerShutdown, "daemon is shutting down"} | |||
| 280 | + | } | |||
| 281 | + | return nil, nil, false, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | |||
| 282 | + | } | |||
| 283 | + | _ = ac.closeCapturedTransport(oldSnap.transport) | |||
| 284 | + | d.log.Info("live resume credential parked for reconnect", "session", sess.name) | |||
| 285 | + | return d.resumeParked(h, tr, sz) | |||
| 286 | + | } | |||
| 287 | + | ||||
| 208 | // resumeParked acquires the parked client's output lock before the daemon | 288 | // resumeParked acquires the parked client's output lock before the daemon | ||
| 209 | // registry lock, preserving the global sendMu > Daemon.mu ordering. The parked | 289 | // registry lock, preserving the global sendMu > Daemon.mu ordering. The parked | ||
| 210 | // entry is revalidated after both locks are held because it may expire between | 290 | // entry is revalidated after both locks are held because it may expire between | ||
| @@ -218,6 +298,9 @@ func (d *Daemon) resumeParked(h ports.Hello, tr ports.Transport, sz domain.Size) | |||||
| 218 | } | 298 | } | ||
| 219 | 299 | ||||
| 220 | ac := parked.ac | 300 | ac := parked.ac | ||
| 301 | + | if d.beforeResumeParkedSendMu != nil { | |||
| 302 | + | d.beforeResumeParkedSendMu() | |||
| 303 | + | } | |||
| 221 | ac.sendMu.Lock() | 304 | ac.sendMu.Lock() | ||
| 222 | defer ac.sendMu.Unlock() | 305 | defer ac.sendMu.Unlock() | ||
| 223 | d.mu.Lock() | 306 | d.mu.Lock() | ||
M internal/usecase/daemon/resume_test.go +287 -0
| @@ -268,6 +268,293 @@ func TestEphemeralLinkLossParksAndResumes(t *testing.T) { | |||
| 268 | 268 | require.Same(t, resumedAC, sess.client) | |
| 269 | 269 | } | |
| 270 | 270 | ||
| 271 | + | // TestResumeParkedTokenReplacedDuringWaitFailsClosed covers the lifecycle race | |
| 272 | + | // where a parked credential is consumed or replaced while IntentResume waits on | |
| 273 | + | // the attachment send lock. The nonzero token must fail closed instead of | |
| 274 | + | // falling through to ordinary attach/create routing. | |
| 275 | + | func TestResumeParkedTokenReplacedDuringWaitFailsClosed(t *testing.T) { | |
| 276 | + | pty, release := newBlockingPTY(t) | |
| 277 | + | defer release() | |
| 278 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| 279 | + | ||
| 280 | + | oldTr := &closeTrackingTransport{} | |
| 281 | + | sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr) | |
| 282 | + | require.NoError(t, err) | |
| 283 | + | token := ac.resumeToken | |
| 284 | + | require.NotZero(t, token) | |
| 285 | + | d.clientGone(sess, ac, oldTr, false) | |
| 286 | + | require.Nil(t, sess.client) | |
| 287 | + | d.mu.Lock() | |
| 288 | + | parked := d.parked[token] | |
| 289 | + | sessionsBefore := len(d.sessions) | |
| 290 | + | d.mu.Unlock() | |
| 291 | + | require.NotNil(t, parked) | |
| 292 | + | ||
| 293 | + | reachedLookup := make(chan struct{}) | |
| 294 | + | releaseLookup := make(chan struct{}) | |
| 295 | + | var releaseOnce sync.Once | |
| 296 | + | t.Cleanup(func() { releaseOnce.Do(func() { close(releaseLookup) }) }) | |
| 297 | + | d.beforeResumeParkedSendMu = func() { | |
| 298 | + | close(reachedLookup) | |
| 299 | + | <-releaseLookup | |
| 300 | + | } | |
| 301 | + | ||
| 302 | + | resumeTr := &closeTrackingTransport{} | |
| 303 | + | type routeResult struct { | |
| 304 | + | sess *session | |
| 305 | + | ac *attachedClient | |
| 306 | + | err error | |
| 307 | + | } | |
| 308 | + | result := make(chan routeResult, 1) | |
| 309 | + | go func() { | |
| 310 | + | routedSess, routedAC, routeErr := d.route(helloResumeCapable(ports.IntentResume, "work", token), resumeTr) | |
| 311 | + | result <- routeResult{sess: routedSess, ac: routedAC, err: routeErr} | |
| 312 | + | }() | |
| 313 | + | awaitTestCompletion(t, reachedLookup, "resumeParked did not pause after the parked lookup") | |
| 314 | + | ||
| 315 | + | d.mu.Lock() | |
| 316 | + | require.Same(t, parked, d.parked[token], "fixture: token must still be parked at the seam") | |
| 317 | + | d.removeParkedLocked(token, parked) | |
| 318 | + | require.Nil(t, d.parked[token]) | |
| 319 | + | require.Same(t, sess, d.sessions[sess.id], "fixture: named session must remain registered") | |
| 320 | + | d.mu.Unlock() | |
| 321 | + | ||
| 322 | + | releaseOnce.Do(func() { close(releaseLookup) }) | |
| 323 | + | got := awaitTestValue(t, result, "IntentResume did not finish after parked token replacement") | |
| 324 | + | ||
| 325 | + | require.Error(t, got.err) | |
| 326 | + | require.Nil(t, got.sess) | |
| 327 | + | require.Nil(t, got.ac) | |
| 328 | + | var pe *protoErr | |
| 329 | + | require.ErrorAs(t, got.err, &pe) | |
| 330 | + | require.Equal(t, ports.ErrNoSuchSession, pe.code) | |
| 331 | + | require.Contains(t, pe.Error(), "resume token is no longer valid") | |
| 332 | + | require.Nil(t, sess.client, "lifecycle-race resume must not take over the named session") | |
| 333 | + | require.Empty(t, resumeTr.Sends(), "failed resume must not complete a Welcome handshake") | |
| 334 | + | d.mu.Lock() | |
| 335 | + | _, stillParked := d.parked[token] | |
| 336 | + | sessionsAfter := len(d.sessions) | |
| 337 | + | d.mu.Unlock() | |
| 338 | + | require.False(t, stillParked, "consumed token must stay invalid") | |
| 339 | + | require.Equal(t, sessionsBefore, sessionsAfter, "lifecycle-race resume must not create a session") | |
| 340 | + | } | |
| 341 | + | ||
| 342 | + | // TestResumeLiveAttachmentParkedResumeRaceFailsClosed covers the live-recovery | |
| 343 | + | // path that parks a still-active credential and then loses a competing | |
| 344 | + | // resumeParked race before sendMu revalidation. The lifecycle sentinel must | |
| 345 | + | // surface as fail-closed ErrNoSuchSession, never as a raw ErrInternal leak to | |
| 346 | + | // handleHello. | |
| 347 | + | func TestResumeLiveAttachmentParkedResumeRaceFailsClosed(t *testing.T) { | |
| 348 | + | pty, release := newBlockingPTY(t) | |
| 349 | + | defer release() | |
| 350 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| 351 | + | ||
| 352 | + | oldTr := &closeTrackingTransport{} | |
| 353 | + | sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr) | |
| 354 | + | require.NoError(t, err) | |
| 355 | + | token := ac.resumeToken | |
| 356 | + | require.NotZero(t, token) | |
| 357 | + | require.Same(t, ac, sess.client) | |
| 358 | + | d.mu.Lock() | |
| 359 | + | _, parkedAtStart := d.parked[token] | |
| 360 | + | sessionsBefore := len(d.sessions) | |
| 361 | + | d.mu.Unlock() | |
| 362 | + | require.False(t, parkedAtStart, "fixture: live attachment must not be parked yet") | |
| 363 | + | ||
| 364 | + | reachedLookup := make(chan struct{}) | |
| 365 | + | releaseLookup := make(chan struct{}) | |
| 366 | + | var releaseOnce sync.Once | |
| 367 | + | t.Cleanup(func() { releaseOnce.Do(func() { close(releaseLookup) }) }) | |
| 368 | + | d.beforeResumeParkedSendMu = func() { | |
| 369 | + | close(reachedLookup) | |
| 370 | + | <-releaseLookup | |
| 371 | + | } | |
| 372 | + | ||
| 373 | + | resumeTr := &closeTrackingTransport{} | |
| 374 | + | type routeResult struct { | |
| 375 | + | sess *session | |
| 376 | + | ac *attachedClient | |
| 377 | + | err error | |
| 378 | + | } | |
| 379 | + | result := make(chan routeResult, 1) | |
| 380 | + | go func() { | |
| 381 | + | routedSess, routedAC, routeErr := d.route(helloResumeCapable(ports.IntentResume, "work", token), resumeTr) | |
| 382 | + | result <- routeResult{sess: routedSess, ac: routedAC, err: routeErr} | |
| 383 | + | }() | |
| 384 | + | awaitTestCompletion(t, reachedLookup, "live resume did not pause inside resumeParked after parking") | |
| 385 | + | ||
| 386 | + | d.mu.Lock() | |
| 387 | + | parked := d.parked[token] | |
| 388 | + | require.NotNil(t, parked, "fixture: live recovery must have parked before the sendMu seam") | |
| 389 | + | require.Same(t, ac, parked.ac) | |
| 390 | + | d.removeParkedLocked(token, parked) | |
| 391 | + | require.Nil(t, d.parked[token]) | |
| 392 | + | require.Same(t, sess, d.sessions[sess.id], "fixture: named session must remain registered") | |
| 393 | + | d.mu.Unlock() | |
| 394 | + | ||
| 395 | + | releaseOnce.Do(func() { close(releaseLookup) }) | |
| 396 | + | got := awaitTestValue(t, result, "live IntentResume did not finish after competing resume consumed the token") | |
| 397 | + | ||
| 398 | + | require.Error(t, got.err) | |
| 399 | + | require.Nil(t, got.sess) | |
| 400 | + | require.Nil(t, got.ac) | |
| 401 | + | require.False(t, errors.Is(got.err, errResumeTokenLifecycleRace), "raw lifecycle sentinel must not reach route callers") | |
| 402 | + | var pe *protoErr | |
| 403 | + | require.ErrorAs(t, got.err, &pe) | |
| 404 | + | require.Equal(t, ports.ErrNoSuchSession, pe.code) | |
| 405 | + | require.Contains(t, pe.Error(), "resume token is no longer valid") | |
| 406 | + | require.Nil(t, sess.client, "losing live resume must not take over the named session") | |
| 407 | + | require.Empty(t, resumeTr.Sends(), "failed live resume must not complete a Welcome handshake") | |
| 408 | + | d.mu.Lock() | |
| 409 | + | _, stillParked := d.parked[token] | |
| 410 | + | sessionsAfter := len(d.sessions) | |
| 411 | + | d.mu.Unlock() | |
| 412 | + | require.False(t, stillParked, "consumed token must stay invalid") | |
| 413 | + | require.Equal(t, sessionsBefore, sessionsAfter, "losing live resume must not create a session") | |
| 414 | + | } | |
| 415 | + | ||
| 416 | + | // TestResumeActiveTokenBeforeParkRecoversSameAttachment covers the laptop-sleep | |
| 417 | + | // reconnect race: IntentResume arrives with the live attachment's token before | |
| 418 | + | // transport teardown has parked it. Same client must recover; unknown and | |
| 419 | + | // mismatched credentials stay fail-closed. | |
| 420 | + | func TestResumeActiveTokenBeforeParkRecoversSameAttachment(t *testing.T) { | |
| 421 | + | pty, release := newBlockingPTY(t) | |
| 422 | + | defer release() | |
| 423 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| 424 | + | ||
| 425 | + | oldTr := &closeTrackingTransport{} | |
| 426 | + | sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr) | |
| 427 | + | require.NoError(t, err) | |
| 428 | + | token := ac.resumeToken | |
| 429 | + | require.NotZero(t, token) | |
| 430 | + | require.True(t, ac.resumeCapable) | |
| 431 | + | require.Same(t, ac, sess.client) | |
| 432 | + | d.mu.Lock() | |
| 433 | + | _, parked := d.parked[token] | |
| 434 | + | d.mu.Unlock() | |
| 435 | + | require.False(t, parked, "fixture: live attachment must not be parked yet") | |
| 436 | + | ||
| 437 | + | _, _, err = d.route(helloResumeCapable(ports.IntentResume, "work", 0xdecafbad), &closeTrackingTransport{}) | |
| 438 | + | require.Error(t, err) | |
| 439 | + | var unknownErr *protoErr | |
| 440 | + | require.ErrorAs(t, err, &unknownErr) | |
| 441 | + | require.Equal(t, ports.ErrNoSuchSession, unknownErr.code) | |
| 442 | + | require.Contains(t, unknownErr.Error(), "resume token is no longer valid") | |
| 443 | + | require.Same(t, ac, sess.client) | |
| 444 | + | require.Equal(t, token, ac.resumeToken) | |
| 445 | + | require.Same(t, oldTr, ac.transport()) | |
| 446 | + | ||
| 447 | + | wrongClient := helloResumeCapable(ports.IntentResume, "work", token) | |
| 448 | + | wrongClient.ClientID = [16]byte{9, 9, 9, 9} | |
| 449 | + | _, _, err = d.route(wrongClient, &closeTrackingTransport{}) | |
| 450 | + | require.Error(t, err) | |
| 451 | + | var mismatchErr *protoErr | |
| 452 | + | require.ErrorAs(t, err, &mismatchErr) | |
| 453 | + | require.Equal(t, ports.ErrNoSuchSession, mismatchErr.code) | |
| 454 | + | require.Contains(t, mismatchErr.Error(), "resume token is no longer valid") | |
| 455 | + | require.Same(t, ac, sess.client) | |
| 456 | + | require.Equal(t, token, ac.resumeToken) | |
| 457 | + | require.Same(t, oldTr, ac.transport()) | |
| 458 | + | ||
| 459 | + | newTr := &closeTrackingTransport{} | |
| 460 | + | resumedSess, resumedAC, err := d.route(helloResumeCapable(ports.IntentResume, "work", token), newTr) | |
| 461 | + | require.NoError(t, err, "same-token same-client IntentResume must recover before parking") | |
| 462 | + | require.Same(t, sess, resumedSess) | |
| 463 | + | require.Same(t, ac, resumedAC) | |
| 464 | + | require.Same(t, ac, sess.client) | |
| 465 | + | require.Same(t, newTr, ac.transport()) | |
| 466 | + | require.NotEqual(t, token, ac.resumeToken, "successful resume rotates the credential") | |
| 467 | + | d.mu.Lock() | |
| 468 | + | _, oldTokenParked := d.parked[token] | |
| 469 | + | d.mu.Unlock() | |
| 470 | + | require.False(t, oldTokenParked, "consumed live credential must not remain resumeable") | |
| 471 | + | ||
| 472 | + | d.clientGone(sess, ac, oldTr, false) | |
| 473 | + | require.Same(t, ac, sess.client, "retired pre-park transport must not detach the rebound attachment") | |
| 474 | + | require.Same(t, newTr, ac.transport()) | |
| 475 | + | require.False(t, newTr.Closed()) | |
| 476 | + | } | |
| 477 | + | ||
| 478 | + | // TestStaleClientGoneAfterTransportCheckDoesNotDetachReboundAttachment covers | |
| 479 | + | // the TOCTOU where clientGone passes the old-transport precheck, then a live | |
| 480 | + | // resume rebinds the same attachment before detach runs. | |
| 481 | + | func TestStaleClientGoneAfterTransportCheckDoesNotDetachReboundAttachment(t *testing.T) { | |
| 482 | + | pty, release := newBlockingPTY(t) | |
| 483 | + | defer release() | |
| 484 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| 485 | + | ||
| 486 | + | oldTr := &closeTrackingTransport{} | |
| 487 | + | sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr) | |
| 488 | + | require.NoError(t, err) | |
| 489 | + | token := ac.resumeToken | |
| 490 | + | require.NotZero(t, token) | |
| 491 | + | ||
| 492 | + | passedPrecheck := make(chan struct{}) | |
| 493 | + | releaseDetach := make(chan struct{}) | |
| 494 | + | var releaseOnce sync.Once | |
| 495 | + | t.Cleanup(func() { releaseOnce.Do(func() { close(releaseDetach) }) }) | |
| 496 | + | d.beforeClientGoneDetach = func() { | |
| 497 | + | close(passedPrecheck) | |
| 498 | + | <-releaseDetach | |
| 499 | + | } | |
| 500 | + | ||
| 501 | + | goneDone := make(chan struct{}) | |
| 502 | + | go func() { | |
| 503 | + | d.clientGone(sess, ac, oldTr, false) | |
| 504 | + | close(goneDone) | |
| 505 | + | }() | |
| 506 | + | awaitTestCompletion(t, passedPrecheck, "stale clientGone did not reach the post-precheck seam") | |
| 507 | + | ||
| 508 | + | newTr := &closeTrackingTransport{} | |
| 509 | + | resumedSess, resumedAC, err := d.route(helloResumeCapable(ports.IntentResume, "work", token), newTr) | |
| 510 | + | require.NoError(t, err) | |
| 511 | + | require.Same(t, sess, resumedSess) | |
| 512 | + | require.Same(t, ac, resumedAC) | |
| 513 | + | require.Same(t, newTr, ac.transport()) | |
| 514 | + | ||
| 515 | + | releaseOnce.Do(func() { close(releaseDetach) }) | |
| 516 | + | awaitTestCompletion(t, goneDone, "stale clientGone did not finish after resume rebound") | |
| 517 | + | ||
| 518 | + | require.Same(t, ac, sess.client, "stale cleanup must not detach the rebound attachment") | |
| 519 | + | require.Same(t, newTr, ac.transport()) | |
| 520 | + | require.False(t, newTr.Closed(), "rebound transport must survive stale old-link cleanup") | |
| 521 | + | require.True(t, oldTr.Closed(), "live resume retires the captured old transport") | |
| 522 | + | } | |
| 523 | + | ||
| 524 | + | // TestResumeLiveAttachmentParkFailureRetiresOldTransport covers shutdown racing | |
| 525 | + | // park after live detach: the old link must still be revoked/closed exactly once | |
| 526 | + | // and the session must not keep a half-detached owner. | |
| 527 | + | func TestResumeLiveAttachmentParkFailureRetiresOldTransport(t *testing.T) { | |
| 528 | + | pty, release := newBlockingPTY(t) | |
| 529 | + | defer release() | |
| 530 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| 531 | + | ||
| 532 | + | oldTr := &closeTrackingTransport{} | |
| 533 | + | sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr) | |
| 534 | + | require.NoError(t, err) | |
| 535 | + | token := ac.resumeToken | |
| 536 | + | require.NotZero(t, token) | |
| 537 | + | ||
| 538 | + | d.afterDetachRoleEffectsFrozen = func() { | |
| 539 | + | d.mu.Lock() | |
| 540 | + | d.closing = true | |
| 541 | + | d.mu.Unlock() | |
| 542 | + | } | |
| 543 | + | ||
| 544 | + | _, _, err = d.route(helloResumeCapable(ports.IntentResume, "work", token), &closeTrackingTransport{}) | |
| 545 | + | require.Error(t, err) | |
| 546 | + | var pe *protoErr | |
| 547 | + | require.ErrorAs(t, err, &pe) | |
| 548 | + | require.Equal(t, ports.ErrServerShutdown, pe.code) | |
| 549 | + | require.Nil(t, sess.client, "failed live park must leave the session without an owner") | |
| 550 | + | require.Nil(t, ac.transport(), "failed live park must revoke the captured old transport") | |
| 551 | + | require.True(t, oldTr.Closed(), "failed live park must close the captured old transport") | |
| 552 | + | d.mu.Lock() | |
| 553 | + | _, parked := d.parked[token] | |
| 554 | + | d.mu.Unlock() | |
| 555 | + | require.False(t, parked, "failed live park must not publish a resume credential") | |
| 556 | + | } | |
| 557 | + | ||
| 271 | 558 | func TestResumeRebindsRotatesAndDoesNotOpenPTY(t *testing.T) { | |
| 272 | 559 | pty, release := newBlockingPTY(t) | |
| 273 | 560 | defer release() | |
| @@ -268,6 +268,293 @@ func TestEphemeralLinkLossParksAndResumes(t *testing.T) { | |||||
| 268 | require.Same(t, resumedAC, sess.client) | 268 | require.Same(t, resumedAC, sess.client) | ||
| 269 | } | 269 | } | ||
| 270 | 270 | ||||
| 271 | + | // TestResumeParkedTokenReplacedDuringWaitFailsClosed covers the lifecycle race | |||
| 272 | + | // where a parked credential is consumed or replaced while IntentResume waits on | |||
| 273 | + | // the attachment send lock. The nonzero token must fail closed instead of | |||
| 274 | + | // falling through to ordinary attach/create routing. | |||
| 275 | + | func TestResumeParkedTokenReplacedDuringWaitFailsClosed(t *testing.T) { | |||
| 276 | + | pty, release := newBlockingPTY(t) | |||
| 277 | + | defer release() | |||
| 278 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |||
| 279 | + | ||||
| 280 | + | oldTr := &closeTrackingTransport{} | |||
| 281 | + | sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr) | |||
| 282 | + | require.NoError(t, err) | |||
| 283 | + | token := ac.resumeToken | |||
| 284 | + | require.NotZero(t, token) | |||
| 285 | + | d.clientGone(sess, ac, oldTr, false) | |||
| 286 | + | require.Nil(t, sess.client) | |||
| 287 | + | d.mu.Lock() | |||
| 288 | + | parked := d.parked[token] | |||
| 289 | + | sessionsBefore := len(d.sessions) | |||
| 290 | + | d.mu.Unlock() | |||
| 291 | + | require.NotNil(t, parked) | |||
| 292 | + | ||||
| 293 | + | reachedLookup := make(chan struct{}) | |||
| 294 | + | releaseLookup := make(chan struct{}) | |||
| 295 | + | var releaseOnce sync.Once | |||
| 296 | + | t.Cleanup(func() { releaseOnce.Do(func() { close(releaseLookup) }) }) | |||
| 297 | + | d.beforeResumeParkedSendMu = func() { | |||
| 298 | + | close(reachedLookup) | |||
| 299 | + | <-releaseLookup | |||
| 300 | + | } | |||
| 301 | + | ||||
| 302 | + | resumeTr := &closeTrackingTransport{} | |||
| 303 | + | type routeResult struct { | |||
| 304 | + | sess *session | |||
| 305 | + | ac *attachedClient | |||
| 306 | + | err error | |||
| 307 | + | } | |||
| 308 | + | result := make(chan routeResult, 1) | |||
| 309 | + | go func() { | |||
| 310 | + | routedSess, routedAC, routeErr := d.route(helloResumeCapable(ports.IntentResume, "work", token), resumeTr) | |||
| 311 | + | result <- routeResult{sess: routedSess, ac: routedAC, err: routeErr} | |||
| 312 | + | }() | |||
| 313 | + | awaitTestCompletion(t, reachedLookup, "resumeParked did not pause after the parked lookup") | |||
| 314 | + | ||||
| 315 | + | d.mu.Lock() | |||
| 316 | + | require.Same(t, parked, d.parked[token], "fixture: token must still be parked at the seam") | |||
| 317 | + | d.removeParkedLocked(token, parked) | |||
| 318 | + | require.Nil(t, d.parked[token]) | |||
| 319 | + | require.Same(t, sess, d.sessions[sess.id], "fixture: named session must remain registered") | |||
| 320 | + | d.mu.Unlock() | |||
| 321 | + | ||||
| 322 | + | releaseOnce.Do(func() { close(releaseLookup) }) | |||
| 323 | + | got := awaitTestValue(t, result, "IntentResume did not finish after parked token replacement") | |||
| 324 | + | ||||
| 325 | + | require.Error(t, got.err) | |||
| 326 | + | require.Nil(t, got.sess) | |||
| 327 | + | require.Nil(t, got.ac) | |||
| 328 | + | var pe *protoErr | |||
| 329 | + | require.ErrorAs(t, got.err, &pe) | |||
| 330 | + | require.Equal(t, ports.ErrNoSuchSession, pe.code) | |||
| 331 | + | require.Contains(t, pe.Error(), "resume token is no longer valid") | |||
| 332 | + | require.Nil(t, sess.client, "lifecycle-race resume must not take over the named session") | |||
| 333 | + | require.Empty(t, resumeTr.Sends(), "failed resume must not complete a Welcome handshake") | |||
| 334 | + | d.mu.Lock() | |||
| 335 | + | _, stillParked := d.parked[token] | |||
| 336 | + | sessionsAfter := len(d.sessions) | |||
| 337 | + | d.mu.Unlock() | |||
| 338 | + | require.False(t, stillParked, "consumed token must stay invalid") | |||
| 339 | + | require.Equal(t, sessionsBefore, sessionsAfter, "lifecycle-race resume must not create a session") | |||
| 340 | + | } | |||
| 341 | + | ||||
| 342 | + | // TestResumeLiveAttachmentParkedResumeRaceFailsClosed covers the live-recovery | |||
| 343 | + | // path that parks a still-active credential and then loses a competing | |||
| 344 | + | // resumeParked race before sendMu revalidation. The lifecycle sentinel must | |||
| 345 | + | // surface as fail-closed ErrNoSuchSession, never as a raw ErrInternal leak to | |||
| 346 | + | // handleHello. | |||
| 347 | + | func TestResumeLiveAttachmentParkedResumeRaceFailsClosed(t *testing.T) { | |||
| 348 | + | pty, release := newBlockingPTY(t) | |||
| 349 | + | defer release() | |||
| 350 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |||
| 351 | + | ||||
| 352 | + | oldTr := &closeTrackingTransport{} | |||
| 353 | + | sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr) | |||
| 354 | + | require.NoError(t, err) | |||
| 355 | + | token := ac.resumeToken | |||
| 356 | + | require.NotZero(t, token) | |||
| 357 | + | require.Same(t, ac, sess.client) | |||
| 358 | + | d.mu.Lock() | |||
| 359 | + | _, parkedAtStart := d.parked[token] | |||
| 360 | + | sessionsBefore := len(d.sessions) | |||
| 361 | + | d.mu.Unlock() | |||
| 362 | + | require.False(t, parkedAtStart, "fixture: live attachment must not be parked yet") | |||
| 363 | + | ||||
| 364 | + | reachedLookup := make(chan struct{}) | |||
| 365 | + | releaseLookup := make(chan struct{}) | |||
| 366 | + | var releaseOnce sync.Once | |||
| 367 | + | t.Cleanup(func() { releaseOnce.Do(func() { close(releaseLookup) }) }) | |||
| 368 | + | d.beforeResumeParkedSendMu = func() { | |||
| 369 | + | close(reachedLookup) | |||
| 370 | + | <-releaseLookup | |||
| 371 | + | } | |||
| 372 | + | ||||
| 373 | + | resumeTr := &closeTrackingTransport{} | |||
| 374 | + | type routeResult struct { | |||
| 375 | + | sess *session | |||
| 376 | + | ac *attachedClient | |||
| 377 | + | err error | |||
| 378 | + | } | |||
| 379 | + | result := make(chan routeResult, 1) | |||
| 380 | + | go func() { | |||
| 381 | + | routedSess, routedAC, routeErr := d.route(helloResumeCapable(ports.IntentResume, "work", token), resumeTr) | |||
| 382 | + | result <- routeResult{sess: routedSess, ac: routedAC, err: routeErr} | |||
| 383 | + | }() | |||
| 384 | + | awaitTestCompletion(t, reachedLookup, "live resume did not pause inside resumeParked after parking") | |||
| 385 | + | ||||
| 386 | + | d.mu.Lock() | |||
| 387 | + | parked := d.parked[token] | |||
| 388 | + | require.NotNil(t, parked, "fixture: live recovery must have parked before the sendMu seam") | |||
| 389 | + | require.Same(t, ac, parked.ac) | |||
| 390 | + | d.removeParkedLocked(token, parked) | |||
| 391 | + | require.Nil(t, d.parked[token]) | |||
| 392 | + | require.Same(t, sess, d.sessions[sess.id], "fixture: named session must remain registered") | |||
| 393 | + | d.mu.Unlock() | |||
| 394 | + | ||||
| 395 | + | releaseOnce.Do(func() { close(releaseLookup) }) | |||
| 396 | + | got := awaitTestValue(t, result, "live IntentResume did not finish after competing resume consumed the token") | |||
| 397 | + | ||||
| 398 | + | require.Error(t, got.err) | |||
| 399 | + | require.Nil(t, got.sess) | |||
| 400 | + | require.Nil(t, got.ac) | |||
| 401 | + | require.False(t, errors.Is(got.err, errResumeTokenLifecycleRace), "raw lifecycle sentinel must not reach route callers") | |||
| 402 | + | var pe *protoErr | |||
| 403 | + | require.ErrorAs(t, got.err, &pe) | |||
| 404 | + | require.Equal(t, ports.ErrNoSuchSession, pe.code) | |||
| 405 | + | require.Contains(t, pe.Error(), "resume token is no longer valid") | |||
| 406 | + | require.Nil(t, sess.client, "losing live resume must not take over the named session") | |||
| 407 | + | require.Empty(t, resumeTr.Sends(), "failed live resume must not complete a Welcome handshake") | |||
| 408 | + | d.mu.Lock() | |||
| 409 | + | _, stillParked := d.parked[token] | |||
| 410 | + | sessionsAfter := len(d.sessions) | |||
| 411 | + | d.mu.Unlock() | |||
| 412 | + | require.False(t, stillParked, "consumed token must stay invalid") | |||
| 413 | + | require.Equal(t, sessionsBefore, sessionsAfter, "losing live resume must not create a session") | |||
| 414 | + | } | |||
| 415 | + | ||||
| 416 | + | // TestResumeActiveTokenBeforeParkRecoversSameAttachment covers the laptop-sleep | |||
| 417 | + | // reconnect race: IntentResume arrives with the live attachment's token before | |||
| 418 | + | // transport teardown has parked it. Same client must recover; unknown and | |||
| 419 | + | // mismatched credentials stay fail-closed. | |||
| 420 | + | func TestResumeActiveTokenBeforeParkRecoversSameAttachment(t *testing.T) { | |||
| 421 | + | pty, release := newBlockingPTY(t) | |||
| 422 | + | defer release() | |||
| 423 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |||
| 424 | + | ||||
| 425 | + | oldTr := &closeTrackingTransport{} | |||
| 426 | + | sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr) | |||
| 427 | + | require.NoError(t, err) | |||
| 428 | + | token := ac.resumeToken | |||
| 429 | + | require.NotZero(t, token) | |||
| 430 | + | require.True(t, ac.resumeCapable) | |||
| 431 | + | require.Same(t, ac, sess.client) | |||
| 432 | + | d.mu.Lock() | |||
| 433 | + | _, parked := d.parked[token] | |||
| 434 | + | d.mu.Unlock() | |||
| 435 | + | require.False(t, parked, "fixture: live attachment must not be parked yet") | |||
| 436 | + | ||||
| 437 | + | _, _, err = d.route(helloResumeCapable(ports.IntentResume, "work", 0xdecafbad), &closeTrackingTransport{}) | |||
| 438 | + | require.Error(t, err) | |||
| 439 | + | var unknownErr *protoErr | |||
| 440 | + | require.ErrorAs(t, err, &unknownErr) | |||
| 441 | + | require.Equal(t, ports.ErrNoSuchSession, unknownErr.code) | |||
| 442 | + | require.Contains(t, unknownErr.Error(), "resume token is no longer valid") | |||
| 443 | + | require.Same(t, ac, sess.client) | |||
| 444 | + | require.Equal(t, token, ac.resumeToken) | |||
| 445 | + | require.Same(t, oldTr, ac.transport()) | |||
| 446 | + | ||||
| 447 | + | wrongClient := helloResumeCapable(ports.IntentResume, "work", token) | |||
| 448 | + | wrongClient.ClientID = [16]byte{9, 9, 9, 9} | |||
| 449 | + | _, _, err = d.route(wrongClient, &closeTrackingTransport{}) | |||
| 450 | + | require.Error(t, err) | |||
| 451 | + | var mismatchErr *protoErr | |||
| 452 | + | require.ErrorAs(t, err, &mismatchErr) | |||
| 453 | + | require.Equal(t, ports.ErrNoSuchSession, mismatchErr.code) | |||
| 454 | + | require.Contains(t, mismatchErr.Error(), "resume token is no longer valid") | |||
| 455 | + | require.Same(t, ac, sess.client) | |||
| 456 | + | require.Equal(t, token, ac.resumeToken) | |||
| 457 | + | require.Same(t, oldTr, ac.transport()) | |||
| 458 | + | ||||
| 459 | + | newTr := &closeTrackingTransport{} | |||
| 460 | + | resumedSess, resumedAC, err := d.route(helloResumeCapable(ports.IntentResume, "work", token), newTr) | |||
| 461 | + | require.NoError(t, err, "same-token same-client IntentResume must recover before parking") | |||
| 462 | + | require.Same(t, sess, resumedSess) | |||
| 463 | + | require.Same(t, ac, resumedAC) | |||
| 464 | + | require.Same(t, ac, sess.client) | |||
| 465 | + | require.Same(t, newTr, ac.transport()) | |||
| 466 | + | require.NotEqual(t, token, ac.resumeToken, "successful resume rotates the credential") | |||
| 467 | + | d.mu.Lock() | |||
| 468 | + | _, oldTokenParked := d.parked[token] | |||
| 469 | + | d.mu.Unlock() | |||
| 470 | + | require.False(t, oldTokenParked, "consumed live credential must not remain resumeable") | |||
| 471 | + | ||||
| 472 | + | d.clientGone(sess, ac, oldTr, false) | |||
| 473 | + | require.Same(t, ac, sess.client, "retired pre-park transport must not detach the rebound attachment") | |||
| 474 | + | require.Same(t, newTr, ac.transport()) | |||
| 475 | + | require.False(t, newTr.Closed()) | |||
| 476 | + | } | |||
| 477 | + | ||||
| 478 | + | // TestStaleClientGoneAfterTransportCheckDoesNotDetachReboundAttachment covers | |||
| 479 | + | // the TOCTOU where clientGone passes the old-transport precheck, then a live | |||
| 480 | + | // resume rebinds the same attachment before detach runs. | |||
| 481 | + | func TestStaleClientGoneAfterTransportCheckDoesNotDetachReboundAttachment(t *testing.T) { | |||
| 482 | + | pty, release := newBlockingPTY(t) | |||
| 483 | + | defer release() | |||
| 484 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |||
| 485 | + | ||||
| 486 | + | oldTr := &closeTrackingTransport{} | |||
| 487 | + | sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr) | |||
| 488 | + | require.NoError(t, err) | |||
| 489 | + | token := ac.resumeToken | |||
| 490 | + | require.NotZero(t, token) | |||
| 491 | + | ||||
| 492 | + | passedPrecheck := make(chan struct{}) | |||
| 493 | + | releaseDetach := make(chan struct{}) | |||
| 494 | + | var releaseOnce sync.Once | |||
| 495 | + | t.Cleanup(func() { releaseOnce.Do(func() { close(releaseDetach) }) }) | |||
| 496 | + | d.beforeClientGoneDetach = func() { | |||
| 497 | + | close(passedPrecheck) | |||
| 498 | + | <-releaseDetach | |||
| 499 | + | } | |||
| 500 | + | ||||
| 501 | + | goneDone := make(chan struct{}) | |||
| 502 | + | go func() { | |||
| 503 | + | d.clientGone(sess, ac, oldTr, false) | |||
| 504 | + | close(goneDone) | |||
| 505 | + | }() | |||
| 506 | + | awaitTestCompletion(t, passedPrecheck, "stale clientGone did not reach the post-precheck seam") | |||
| 507 | + | ||||
| 508 | + | newTr := &closeTrackingTransport{} | |||
| 509 | + | resumedSess, resumedAC, err := d.route(helloResumeCapable(ports.IntentResume, "work", token), newTr) | |||
| 510 | + | require.NoError(t, err) | |||
| 511 | + | require.Same(t, sess, resumedSess) | |||
| 512 | + | require.Same(t, ac, resumedAC) | |||
| 513 | + | require.Same(t, newTr, ac.transport()) | |||
| 514 | + | ||||
| 515 | + | releaseOnce.Do(func() { close(releaseDetach) }) | |||
| 516 | + | awaitTestCompletion(t, goneDone, "stale clientGone did not finish after resume rebound") | |||
| 517 | + | ||||
| 518 | + | require.Same(t, ac, sess.client, "stale cleanup must not detach the rebound attachment") | |||
| 519 | + | require.Same(t, newTr, ac.transport()) | |||
| 520 | + | require.False(t, newTr.Closed(), "rebound transport must survive stale old-link cleanup") | |||
| 521 | + | require.True(t, oldTr.Closed(), "live resume retires the captured old transport") | |||
| 522 | + | } | |||
| 523 | + | ||||
| 524 | + | // TestResumeLiveAttachmentParkFailureRetiresOldTransport covers shutdown racing | |||
| 525 | + | // park after live detach: the old link must still be revoked/closed exactly once | |||
| 526 | + | // and the session must not keep a half-detached owner. | |||
| 527 | + | func TestResumeLiveAttachmentParkFailureRetiresOldTransport(t *testing.T) { | |||
| 528 | + | pty, release := newBlockingPTY(t) | |||
| 529 | + | defer release() | |||
| 530 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |||
| 531 | + | ||||
| 532 | + | oldTr := &closeTrackingTransport{} | |||
| 533 | + | sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr) | |||
| 534 | + | require.NoError(t, err) | |||
| 535 | + | token := ac.resumeToken | |||
| 536 | + | require.NotZero(t, token) | |||
| 537 | + | ||||
| 538 | + | d.afterDetachRoleEffectsFrozen = func() { | |||
| 539 | + | d.mu.Lock() | |||
| 540 | + | d.closing = true | |||
| 541 | + | d.mu.Unlock() | |||
| 542 | + | } | |||
| 543 | + | ||||
| 544 | + | _, _, err = d.route(helloResumeCapable(ports.IntentResume, "work", token), &closeTrackingTransport{}) | |||
| 545 | + | require.Error(t, err) | |||
| 546 | + | var pe *protoErr | |||
| 547 | + | require.ErrorAs(t, err, &pe) | |||
| 548 | + | require.Equal(t, ports.ErrServerShutdown, pe.code) | |||
| 549 | + | require.Nil(t, sess.client, "failed live park must leave the session without an owner") | |||
| 550 | + | require.Nil(t, ac.transport(), "failed live park must revoke the captured old transport") | |||
| 551 | + | require.True(t, oldTr.Closed(), "failed live park must close the captured old transport") | |||
| 552 | + | d.mu.Lock() | |||
| 553 | + | _, parked := d.parked[token] | |||
| 554 | + | d.mu.Unlock() | |||
| 555 | + | require.False(t, parked, "failed live park must not publish a resume credential") | |||
| 556 | + | } | |||
| 557 | + | ||||
| 271 | func TestResumeRebindsRotatesAndDoesNotOpenPTY(t *testing.T) { | 558 | func TestResumeRebindsRotatesAndDoesNotOpenPTY(t *testing.T) { | ||
| 272 | pty, release := newBlockingPTY(t) | 559 | pty, release := newBlockingPTY(t) | ||
| 273 | defer release() | 560 | defer release() | ||
M internal/usecase/daemon/session.go +10 -0
| @@ -675,6 +675,13 @@ func (s *session) detachIfCurrent(ac *attachedClient) bool { | |||
| 675 | 675 | // detachIfCurrent publishes terminal role invalidation through the attachment | |
| 676 | 676 | // gate. Freeze/drain happens before routing or session locks are acquired. | |
| 677 | 677 | func (d *Daemon) detachIfCurrent(sess *session, ac *attachedClient) bool { | |
| 678 | + | return d.detachIfCurrentTransport(sess, ac, transportSnapshot{}) | |
| 679 | + | } | |
| 680 | + | ||
| 681 | + | // detachIfCurrentTransport is detachIfCurrent with an optional exact-link | |
| 682 | + | // fence. When expected.transport is non-nil, a rebound incarnation of the same | |
| 683 | + | // attachment object cannot be detached by a stale connection loop. | |
| 684 | + | func (d *Daemon) detachIfCurrentTransport(sess *session, ac *attachedClient, expected transportSnapshot) bool { | |
| 678 | 685 | if sess == nil || ac == nil { | |
| 679 | 686 | return false | |
| 680 | 687 | } | |
| @@ -686,6 +693,9 @@ func (d *Daemon) detachIfCurrent(sess *session, ac *attachedClient) bool { | |||
| 686 | 693 | d.notices.routingMu.Lock() | |
| 687 | 694 | sess.mu.Lock() | |
| 688 | 695 | current := sess.client == ac && ac.currentSession() == sess | |
| 696 | + | if current && expected.transport != nil && !ac.transportSnapshotCurrent(expected) { | |
| 697 | + | current = false | |
| 698 | + | } | |
| 689 | 699 | if current { | |
| 690 | 700 | sess.client = nil | |
| 691 | 701 | ac.roleGeneration.Add(1) | |
| @@ -675,6 +675,13 @@ func (s *session) detachIfCurrent(ac *attachedClient) bool { | |||||
| 675 | // detachIfCurrent publishes terminal role invalidation through the attachment | 675 | // detachIfCurrent publishes terminal role invalidation through the attachment | ||
| 676 | // gate. Freeze/drain happens before routing or session locks are acquired. | 676 | // gate. Freeze/drain happens before routing or session locks are acquired. | ||
| 677 | func (d *Daemon) detachIfCurrent(sess *session, ac *attachedClient) bool { | 677 | func (d *Daemon) detachIfCurrent(sess *session, ac *attachedClient) bool { | ||
| 678 | + | return d.detachIfCurrentTransport(sess, ac, transportSnapshot{}) | |||
| 679 | + | } | |||
| 680 | + | ||||
| 681 | + | // detachIfCurrentTransport is detachIfCurrent with an optional exact-link | |||
| 682 | + | // fence. When expected.transport is non-nil, a rebound incarnation of the same | |||
| 683 | + | // attachment object cannot be detached by a stale connection loop. | |||
| 684 | + | func (d *Daemon) detachIfCurrentTransport(sess *session, ac *attachedClient, expected transportSnapshot) bool { | |||
| 678 | if sess == nil || ac == nil { | 685 | if sess == nil || ac == nil { | ||
| 679 | return false | 686 | return false | ||
| 680 | } | 687 | } | ||
| @@ -686,6 +693,9 @@ func (d *Daemon) detachIfCurrent(sess *session, ac *attachedClient) bool { | |||||
| 686 | d.notices.routingMu.Lock() | 693 | d.notices.routingMu.Lock() | ||
| 687 | sess.mu.Lock() | 694 | sess.mu.Lock() | ||
| 688 | current := sess.client == ac && ac.currentSession() == sess | 695 | current := sess.client == ac && ac.currentSession() == sess | ||
| 696 | + | if current && expected.transport != nil && !ac.transportSnapshotCurrent(expected) { | |||
| 697 | + | current = false | |||
| 698 | + | } | |||
| 689 | if current { | 699 | if current { | ||
| 690 | sess.client = nil | 700 | sess.client = nil | ||
| 691 | ac.roleGeneration.Add(1) | 701 | ac.roleGeneration.Add(1) | ||