fix(daemon): shutdown/create interlock + unblockable teardown ordering
- route now rejects Hellos once shutdown has begun (closing flag set under d.mu atomically with the registry-empty check in killSession and with shutdownAll's snapshot); new ports.ErrServerShutdown wire code - killSession reordered: ctx cancel, pty.Close and the done signal run unconditionally before the Detached notice, which is now asynchronous and deadline-bounded (boundedSend force-closes a wedged transport) - Serve waits for pending notices before hard-closing connections, runs a defensive kill sweep after conn handlers drain, and joins notices last (mu-guarded completion channels, not a WaitGroup: Add-from-zero would race Wait) - detachOnSendError now kills ephemeral sessions like every other detach path - regression tests: Hello racing last-session-kill is rejected with no leaked child; Serve returns despite a write-wedged client; send error kills ephemeral
- commit
- 22dc71c8307e43e923456f5fa2e8182917b88a29
- author
- brice · · 2026-07-02 12:17 UTC
- parent
- 6377eac
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
3 files · +321 -18
M internal/ports/wire.go +1 -0
| @@ -34,6 +34,7 @@ const ( | |||
| 34 | 34 | ErrVersionMismatch uint16 = 1 | |
| 35 | 35 | ErrNoSuchSession uint16 = 2 | |
| 36 | 36 | ErrNameTaken uint16 = 3 | |
| 37 | + | ErrServerShutdown uint16 = 4 | |
| 37 | 38 | ErrInternal uint16 = 255 | |
| 38 | 39 | ) | |
| 39 | 40 | ||
| @@ -34,6 +34,7 @@ const ( | |||||
| 34 | ErrVersionMismatch uint16 = 1 | 34 | ErrVersionMismatch uint16 = 1 | ||
| 35 | ErrNoSuchSession uint16 = 2 | 35 | ErrNoSuchSession uint16 = 2 | ||
| 36 | ErrNameTaken uint16 = 3 | 36 | ErrNameTaken uint16 = 3 | ||
| 37 | + | ErrServerShutdown uint16 = 4 | |||
| 37 | ErrInternal uint16 = 255 | 38 | ErrInternal uint16 = 255 | ||
| 38 | ) | 39 | ) | ||
| 39 | 40 | ||||
M internal/usecase/daemon/daemon.go +135 -18
| @@ -51,6 +51,12 @@ const debounceInterval = 10 * time.Millisecond | |||
| 51 | 51 | // ptyReadBufSize is the PTY reader's read buffer. | |
| 52 | 52 | const ptyReadBufSize = 32 * 1024 | |
| 53 | 53 | ||
| 54 | + | // detachNotifyTimeout bounds the best-effort Detached notification on the | |
| 55 | + | // detach/kill/shutdown paths: if a wedged client (full kernel send buffer) | |
| 56 | + | // blocks the write for this long, its transport is force-closed, which fails | |
| 57 | + | // the in-flight send. Teardown is never gated on a client draining its socket. | |
| 58 | + | const detachNotifyTimeout = time.Second | |
| 59 | + | ||
| 54 | 60 | // defaultSize is used when a client's Hello carries no valid dimensions. | |
| 55 | 61 | var defaultSize = domain.Size{Cols: 80, Rows: 24} | |
| 56 | 62 | ||
| @@ -60,6 +66,18 @@ type Daemon struct { | |||
| 60 | 66 | mu sync.Mutex | |
| 61 | 67 | sessions map[domain.SessionID]*session | |
| 62 | 68 | nextID uint64 | |
| 69 | + | // closing marks that shutdown has irreversibly begun. It is set under mu, | |
| 70 | + | // atomically with the event that makes shutdown inevitable (the registry | |
| 71 | + | // emptying in killSession, or shutdownAll starting), and checked by route | |
| 72 | + | // under the same mutex — so a Hello racing shutdown can never insert a new | |
| 73 | + | // session that nobody would tear down. | |
| 74 | + | closing bool | |
| 75 | + | // notifies holds one completion channel per in-flight async Detached | |
| 76 | + | // notification (guarded by mu, pruned on insert). Channels rather than a | |
| 77 | + | // WaitGroup: notifications are spawned from arbitrary goroutines while | |
| 78 | + | // Serve may be waiting, and WaitGroup forbids Add-from-zero concurrent | |
| 79 | + | // with Wait. | |
| 80 | + | notifies []chan struct{} | |
| 63 | 81 | ||
| 64 | 82 | ptys ports.PTYFactory | |
| 65 | 83 | clock ports.Clock | |
| @@ -129,6 +147,67 @@ func (ac *attachedClient) send(f ports.Frame) error { | |||
| 129 | 147 | return ac.tr.Send(f) | |
| 130 | 148 | } | |
| 131 | 149 | ||
| 150 | + | // boundedSend sends f to ac with a deadline watchdog: if the send (including | |
| 151 | + | // waiting on sendMu behind a wedged paint) does not complete within | |
| 152 | + | // detachNotifyTimeout, the transport is force-closed, failing the in-flight | |
| 153 | + | // write. Detach/kill/shutdown paths use this so they are never gated on a | |
| 154 | + | // client that has stopped draining its socket. | |
| 155 | + | func (d *Daemon) boundedSend(ac *attachedClient, f ports.Frame) { | |
| 156 | + | timer := d.clock.NewTimer(detachNotifyTimeout) | |
| 157 | + | sent := make(chan struct{}) | |
| 158 | + | go func() { | |
| 159 | + | select { | |
| 160 | + | case <-timer.C(): | |
| 161 | + | _ = ac.tr.Close() | |
| 162 | + | case <-sent: | |
| 163 | + | } | |
| 164 | + | }() | |
| 165 | + | _ = ac.send(f) | |
| 166 | + | timer.Stop() | |
| 167 | + | close(sent) | |
| 168 | + | } | |
| 169 | + | ||
| 170 | + | // notifyDetachedAsync delivers a best-effort Detached notice off the caller's | |
| 171 | + | // goroutine and then closes the transport. Session teardown (killSession) must | |
| 172 | + | // complete regardless of client state, so the notice is both asynchronous and | |
| 173 | + | // deadline-bounded; Serve waits for pending notices before force-closing | |
| 174 | + | // connections so a graceful notice is not raced by the hard close. | |
| 175 | + | func (d *Daemon) notifyDetachedAsync(ac *attachedClient, reason uint8) { | |
| 176 | + | done := make(chan struct{}) | |
| 177 | + | d.mu.Lock() | |
| 178 | + | // Prune completed entries so the slice stays bounded by the number of | |
| 179 | + | // notifications actually in flight. | |
| 180 | + | kept := d.notifies[:0] | |
| 181 | + | for _, c := range d.notifies { | |
| 182 | + | select { | |
| 183 | + | case <-c: | |
| 184 | + | default: | |
| 185 | + | kept = append(kept, c) | |
| 186 | + | } | |
| 187 | + | } | |
| 188 | + | d.notifies = append(kept, done) | |
| 189 | + | d.mu.Unlock() | |
| 190 | + | ||
| 191 | + | go func() { | |
| 192 | + | defer close(done) | |
| 193 | + | d.boundedSend(ac, frameDetached(reason)) | |
| 194 | + | _ = ac.tr.Close() | |
| 195 | + | }() | |
| 196 | + | } | |
| 197 | + | ||
| 198 | + | // waitNotifies blocks until every Detached notification in flight at the time | |
| 199 | + | // of the call has completed. Each is deadline-bounded (boundedSend), so this | |
| 200 | + | // wait is bounded too. | |
| 201 | + | func (d *Daemon) waitNotifies() { | |
| 202 | + | d.mu.Lock() | |
| 203 | + | snapshot := make([]chan struct{}, len(d.notifies)) | |
| 204 | + | copy(snapshot, d.notifies) | |
| 205 | + | d.mu.Unlock() | |
| 206 | + | for _, c := range snapshot { | |
| 207 | + | <-c | |
| 208 | + | } | |
| 209 | + | } | |
| 210 | + | ||
| 132 | 211 | // Option customises a Daemon at construction. | |
| 133 | 212 | type Option func(*Daemon) | |
| 134 | 213 | ||
| @@ -198,22 +277,33 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error { | |||
| 198 | 277 | }() | |
| 199 | 278 | } | |
| 200 | 279 | ||
| 201 | - | // Tear down: first notify any attached clients and close their PTYs and | |
| 202 | - | // transports (killSession), THEN force-close whatever connections are still | |
| 203 | - | // parked (control conns, in-flight handshakes), then cancel the serve | |
| 204 | - | // context and wait for every goroutine. | |
| 280 | + | // Tear down. shutdownAll marks the daemon closing (route now rejects any | |
| 281 | + | // racing Hello) and kills every session: PTYs are closed and contexts | |
| 282 | + | // cancelled unconditionally, with the Detached notices sent asynchronously | |
| 283 | + | // under a deadline. Wait for those notices (bounded — a wedged client's | |
| 284 | + | // transport is force-closed by the deadline) before hard-closing the | |
| 285 | + | // remaining parked connections, so graceful notices are never raced by the | |
| 286 | + | // force-close. Finally drain the conn handlers, run one defensive sweep in | |
| 287 | + | // case any ordering ever leaves a session behind, and join the session | |
| 288 | + | // goroutines (readers unblock via pty.Close, schedulers via ctx cancel). | |
| 205 | 289 | d.shutdownAll(ports.ReasonServerShutdown) | |
| 290 | + | d.waitNotifies() | |
| 206 | 291 | d.hardCancel() | |
| 207 | 292 | d.serveCancel() | |
| 208 | - | d.sessWg.Wait() | |
| 209 | 293 | d.connWg.Wait() | |
| 294 | + | d.shutdownAll(ports.ReasonServerShutdown) | |
| 295 | + | d.sessWg.Wait() | |
| 296 | + | d.waitNotifies() | |
| 210 | 297 | return nil | |
| 211 | 298 | } | |
| 212 | 299 | ||
| 213 | - | // shutdownAll kills every live session, snapshotting under the registry lock so | |
| 214 | - | // killSession (which relocks) never runs while the lock is held. | |
| 300 | + | // shutdownAll marks the daemon closing and kills every live session. Setting | |
| 301 | + | // closing under the same lock as the snapshot guarantees no session can be | |
| 302 | + | // inserted after the snapshot: route rejects once closing is set, and both run | |
| 303 | + | // under d.mu. killSession (which relocks) runs after the lock is released. | |
| 215 | 304 | func (d *Daemon) shutdownAll(reason uint8) { | |
| 216 | 305 | d.mu.Lock() | |
| 306 | + | d.closing = true | |
| 217 | 307 | snapshot := make([]*session, 0, len(d.sessions)) | |
| 218 | 308 | for _, s := range d.sessions { | |
| 219 | 309 | snapshot = append(snapshot, s) | |
| @@ -349,6 +439,14 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||
| 349 | 439 | } | |
| 350 | 440 | ||
| 351 | 441 | d.mu.Lock() | |
| 442 | + | // Shutdown/create interlock: once shutdown has begun (last session removed, | |
| 443 | + | // or shutdownAll started) no new session may be created and no attach may | |
| 444 | + | // proceed — the teardown snapshot has already been (or is being) taken, so | |
| 445 | + | // anything inserted now would leak its PTY and hang Serve. | |
| 446 | + | if d.closing { | |
| 447 | + | d.mu.Unlock() | |
| 448 | + | return nil, nil, &protoErr{ports.ErrServerShutdown, "daemon is shutting down"} | |
| 449 | + | } | |
| 352 | 450 | switch h.Intent { | |
| 353 | 451 | case ports.IntentEphemeral: | |
| 354 | 452 | name := d.allocEphemeralNameLocked() | |
| @@ -438,8 +536,9 @@ func (d *Daemon) attachClient(sess *session, tr ports.Transport, sz domain.Size) | |||
| 438 | 536 | sess.mu.Unlock() | |
| 439 | 537 | ||
| 440 | 538 | if old != nil { | |
| 441 | - | _ = old.send(frameDetached(ports.ReasonDetach)) | |
| 442 | - | _ = old.tr.Close() | |
| 539 | + | // Async + bounded: a dead or wedged old client must not stall the new | |
| 540 | + | // client's handshake. | |
| 541 | + | d.notifyDetachedAsync(old, ports.ReasonDetach) | |
| 443 | 542 | } | |
| 444 | 543 | return ac | |
| 445 | 544 | } | |
| @@ -499,7 +598,11 @@ func (d *Daemon) clientGone(sess *session, ac *attachedClient, explicit bool) { | |||
| 499 | 598 | return // already displaced by a newer client; nothing to do | |
| 500 | 599 | } | |
| 501 | 600 | if explicit { | |
| 502 | - | _ = ac.send(frameDetached(ports.ReasonDetach)) | |
| 601 | + | // Synchronous so the ack is delivered before the transport closes | |
| 602 | + | // (the client is actively awaiting it), but deadline-bounded so a | |
| 603 | + | // wedged client cannot pin this conn handler and hang Serve's | |
| 604 | + | // connWg.Wait. | |
| 605 | + | d.boundedSend(ac, frameDetached(ports.ReasonDetach)) | |
| 503 | 606 | } | |
| 504 | 607 | _ = ac.tr.Close() | |
| 505 | 608 | if sess.ephemeral { | |
| @@ -657,18 +760,27 @@ func (d *Daemon) resize(sess *session, ac *attachedClient, sz domain.Size) { | |||
| 657 | 760 | } | |
| 658 | 761 | } | |
| 659 | 762 | ||
| 660 | - | // detachOnSendError drops a client whose transport failed but keeps the session | |
| 661 | - | // running headless (per plan: output errors never kill the session). | |
| 763 | + | // detachOnSendError drops a client whose transport failed. Like every other | |
| 764 | + | // detach path, losing the client kills an ephemeral session (its lifetime is | |
| 765 | + | // its client's); a named session keeps running headless. | |
| 662 | 766 | func (d *Daemon) detachOnSendError(sess *session, ac *attachedClient) { | |
| 663 | 767 | if sess.detachIfCurrent(ac) { | |
| 664 | 768 | _ = ac.tr.Close() | |
| 665 | 769 | d.log.Warn("detached client after send error", "session", sess.name) | |
| 770 | + | if sess.ephemeral { | |
| 771 | + | d.killSession(sess, ports.ReasonSessionKilled) | |
| 772 | + | } | |
| 666 | 773 | } | |
| 667 | 774 | } | |
| 668 | 775 | ||
| 669 | 776 | // killSession removes a session and tears down its resources. It is | |
| 670 | 777 | // idempotent: only the caller that wins the registry delete acts. When the | |
| 671 | - | // registry empties it signals daemon shutdown. | |
| 778 | + | // registry empties it marks the daemon closing (atomically with the | |
| 779 | + | // empty-check, under d.mu) and signals shutdown. | |
| 780 | + | // | |
| 781 | + | // Teardown ordering matters: context cancel, pty.Close, and the done signal | |
| 782 | + | // run first and unconditionally — never gated behind a client send. The | |
| 783 | + | // Detached notice is best-effort, asynchronous, and deadline-bounded. | |
| 672 | 784 | func (d *Daemon) killSession(sess *session, reason uint8) { | |
| 673 | 785 | d.mu.Lock() | |
| 674 | 786 | if _, ok := d.sessions[sess.id]; !ok { | |
| @@ -677,23 +789,28 @@ func (d *Daemon) killSession(sess *session, reason uint8) { | |||
| 677 | 789 | } | |
| 678 | 790 | delete(d.sessions, sess.id) | |
| 679 | 791 | empty := len(d.sessions) == 0 | |
| 792 | + | if empty { | |
| 793 | + | // Shutdown is now inevitable and irreversible (doneOnce below): stop | |
| 794 | + | // route from inserting new sessions from this instant, while we still | |
| 795 | + | // hold the same lock route checks under. | |
| 796 | + | d.closing = true | |
| 797 | + | } | |
| 680 | 798 | d.mu.Unlock() | |
| 681 | 799 | ||
| 682 | 800 | sess.mu.Lock() | |
| 683 | 801 | ac := sess.client | |
| 684 | 802 | sess.client = nil | |
| 685 | 803 | sess.mu.Unlock() | |
| 686 | - | if ac != nil { | |
| 687 | - | _ = ac.send(frameDetached(reason)) | |
| 688 | - | _ = ac.tr.Close() | |
| 689 | - | } | |
| 690 | 804 | ||
| 691 | 805 | sess.cancel() | |
| 692 | 806 | _ = sess.win.pty.Close() | |
| 693 | - | ||
| 694 | 807 | if empty { | |
| 695 | 808 | d.doneOnce.Do(func() { close(d.done) }) | |
| 696 | 809 | } | |
| 810 | + | ||
| 811 | + | if ac != nil { | |
| 812 | + | d.notifyDetachedAsync(ac, reason) | |
| 813 | + | } | |
| 697 | 814 | } | |
| 698 | 815 | ||
| 699 | 816 | // allocEphemeralNameLocked returns the lowest free decimal name. Caller holds | |
| @@ -51,6 +51,12 @@ const debounceInterval = 10 * time.Millisecond | |||||
| 51 | // ptyReadBufSize is the PTY reader's read buffer. | 51 | // ptyReadBufSize is the PTY reader's read buffer. | ||
| 52 | const ptyReadBufSize = 32 * 1024 | 52 | const ptyReadBufSize = 32 * 1024 | ||
| 53 | 53 | ||||
| 54 | + | // detachNotifyTimeout bounds the best-effort Detached notification on the | |||
| 55 | + | // detach/kill/shutdown paths: if a wedged client (full kernel send buffer) | |||
| 56 | + | // blocks the write for this long, its transport is force-closed, which fails | |||
| 57 | + | // the in-flight send. Teardown is never gated on a client draining its socket. | |||
| 58 | + | const detachNotifyTimeout = time.Second | |||
| 59 | + | ||||
| 54 | // defaultSize is used when a client's Hello carries no valid dimensions. | 60 | // defaultSize is used when a client's Hello carries no valid dimensions. | ||
| 55 | var defaultSize = domain.Size{Cols: 80, Rows: 24} | 61 | var defaultSize = domain.Size{Cols: 80, Rows: 24} | ||
| 56 | 62 | ||||
| @@ -60,6 +66,18 @@ type Daemon struct { | |||||
| 60 | mu sync.Mutex | 66 | mu sync.Mutex | ||
| 61 | sessions map[domain.SessionID]*session | 67 | sessions map[domain.SessionID]*session | ||
| 62 | nextID uint64 | 68 | nextID uint64 | ||
| 69 | + | // closing marks that shutdown has irreversibly begun. It is set under mu, | |||
| 70 | + | // atomically with the event that makes shutdown inevitable (the registry | |||
| 71 | + | // emptying in killSession, or shutdownAll starting), and checked by route | |||
| 72 | + | // under the same mutex — so a Hello racing shutdown can never insert a new | |||
| 73 | + | // session that nobody would tear down. | |||
| 74 | + | closing bool | |||
| 75 | + | // notifies holds one completion channel per in-flight async Detached | |||
| 76 | + | // notification (guarded by mu, pruned on insert). Channels rather than a | |||
| 77 | + | // WaitGroup: notifications are spawned from arbitrary goroutines while | |||
| 78 | + | // Serve may be waiting, and WaitGroup forbids Add-from-zero concurrent | |||
| 79 | + | // with Wait. | |||
| 80 | + | notifies []chan struct{} | |||
| 63 | 81 | ||||
| 64 | ptys ports.PTYFactory | 82 | ptys ports.PTYFactory | ||
| 65 | clock ports.Clock | 83 | clock ports.Clock | ||
| @@ -129,6 +147,67 @@ func (ac *attachedClient) send(f ports.Frame) error { | |||||
| 129 | return ac.tr.Send(f) | 147 | return ac.tr.Send(f) | ||
| 130 | } | 148 | } | ||
| 131 | 149 | ||||
| 150 | + | // boundedSend sends f to ac with a deadline watchdog: if the send (including | |||
| 151 | + | // waiting on sendMu behind a wedged paint) does not complete within | |||
| 152 | + | // detachNotifyTimeout, the transport is force-closed, failing the in-flight | |||
| 153 | + | // write. Detach/kill/shutdown paths use this so they are never gated on a | |||
| 154 | + | // client that has stopped draining its socket. | |||
| 155 | + | func (d *Daemon) boundedSend(ac *attachedClient, f ports.Frame) { | |||
| 156 | + | timer := d.clock.NewTimer(detachNotifyTimeout) | |||
| 157 | + | sent := make(chan struct{}) | |||
| 158 | + | go func() { | |||
| 159 | + | select { | |||
| 160 | + | case <-timer.C(): | |||
| 161 | + | _ = ac.tr.Close() | |||
| 162 | + | case <-sent: | |||
| 163 | + | } | |||
| 164 | + | }() | |||
| 165 | + | _ = ac.send(f) | |||
| 166 | + | timer.Stop() | |||
| 167 | + | close(sent) | |||
| 168 | + | } | |||
| 169 | + | ||||
| 170 | + | // notifyDetachedAsync delivers a best-effort Detached notice off the caller's | |||
| 171 | + | // goroutine and then closes the transport. Session teardown (killSession) must | |||
| 172 | + | // complete regardless of client state, so the notice is both asynchronous and | |||
| 173 | + | // deadline-bounded; Serve waits for pending notices before force-closing | |||
| 174 | + | // connections so a graceful notice is not raced by the hard close. | |||
| 175 | + | func (d *Daemon) notifyDetachedAsync(ac *attachedClient, reason uint8) { | |||
| 176 | + | done := make(chan struct{}) | |||
| 177 | + | d.mu.Lock() | |||
| 178 | + | // Prune completed entries so the slice stays bounded by the number of | |||
| 179 | + | // notifications actually in flight. | |||
| 180 | + | kept := d.notifies[:0] | |||
| 181 | + | for _, c := range d.notifies { | |||
| 182 | + | select { | |||
| 183 | + | case <-c: | |||
| 184 | + | default: | |||
| 185 | + | kept = append(kept, c) | |||
| 186 | + | } | |||
| 187 | + | } | |||
| 188 | + | d.notifies = append(kept, done) | |||
| 189 | + | d.mu.Unlock() | |||
| 190 | + | ||||
| 191 | + | go func() { | |||
| 192 | + | defer close(done) | |||
| 193 | + | d.boundedSend(ac, frameDetached(reason)) | |||
| 194 | + | _ = ac.tr.Close() | |||
| 195 | + | }() | |||
| 196 | + | } | |||
| 197 | + | ||||
| 198 | + | // waitNotifies blocks until every Detached notification in flight at the time | |||
| 199 | + | // of the call has completed. Each is deadline-bounded (boundedSend), so this | |||
| 200 | + | // wait is bounded too. | |||
| 201 | + | func (d *Daemon) waitNotifies() { | |||
| 202 | + | d.mu.Lock() | |||
| 203 | + | snapshot := make([]chan struct{}, len(d.notifies)) | |||
| 204 | + | copy(snapshot, d.notifies) | |||
| 205 | + | d.mu.Unlock() | |||
| 206 | + | for _, c := range snapshot { | |||
| 207 | + | <-c | |||
| 208 | + | } | |||
| 209 | + | } | |||
| 210 | + | ||||
| 132 | // Option customises a Daemon at construction. | 211 | // Option customises a Daemon at construction. | ||
| 133 | type Option func(*Daemon) | 212 | type Option func(*Daemon) | ||
| 134 | 213 | ||||
| @@ -198,22 +277,33 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error { | |||||
| 198 | }() | 277 | }() | ||
| 199 | } | 278 | } | ||
| 200 | 279 | ||||
| 201 | - | // Tear down: first notify any attached clients and close their PTYs and | 280 | + | // Tear down. shutdownAll marks the daemon closing (route now rejects any |
| 202 | - | // transports (killSession), THEN force-close whatever connections are still | 281 | + | // racing Hello) and kills every session: PTYs are closed and contexts |
| 203 | - | // parked (control conns, in-flight handshakes), then cancel the serve | 282 | + | // cancelled unconditionally, with the Detached notices sent asynchronously |
| 204 | - | // context and wait for every goroutine. | 283 | + | // under a deadline. Wait for those notices (bounded — a wedged client's |
| 284 | + | // transport is force-closed by the deadline) before hard-closing the | |||
| 285 | + | // remaining parked connections, so graceful notices are never raced by the | |||
| 286 | + | // force-close. Finally drain the conn handlers, run one defensive sweep in | |||
| 287 | + | // case any ordering ever leaves a session behind, and join the session | |||
| 288 | + | // goroutines (readers unblock via pty.Close, schedulers via ctx cancel). | |||
| 205 | d.shutdownAll(ports.ReasonServerShutdown) | 289 | d.shutdownAll(ports.ReasonServerShutdown) | ||
| 290 | + | d.waitNotifies() | |||
| 206 | d.hardCancel() | 291 | d.hardCancel() | ||
| 207 | d.serveCancel() | 292 | d.serveCancel() | ||
| 208 | - | d.sessWg.Wait() | |||
| 209 | d.connWg.Wait() | 293 | d.connWg.Wait() | ||
| 294 | + | d.shutdownAll(ports.ReasonServerShutdown) | |||
| 295 | + | d.sessWg.Wait() | |||
| 296 | + | d.waitNotifies() | |||
| 210 | return nil | 297 | return nil | ||
| 211 | } | 298 | } | ||
| 212 | 299 | ||||
| 213 | - | // shutdownAll kills every live session, snapshotting under the registry lock so | 300 | + | // shutdownAll marks the daemon closing and kills every live session. Setting |
| 214 | - | // killSession (which relocks) never runs while the lock is held. | 301 | + | // closing under the same lock as the snapshot guarantees no session can be |
| 302 | + | // inserted after the snapshot: route rejects once closing is set, and both run | |||
| 303 | + | // under d.mu. killSession (which relocks) runs after the lock is released. | |||
| 215 | func (d *Daemon) shutdownAll(reason uint8) { | 304 | func (d *Daemon) shutdownAll(reason uint8) { | ||
| 216 | d.mu.Lock() | 305 | d.mu.Lock() | ||
| 306 | + | d.closing = true | |||
| 217 | snapshot := make([]*session, 0, len(d.sessions)) | 307 | snapshot := make([]*session, 0, len(d.sessions)) | ||
| 218 | for _, s := range d.sessions { | 308 | for _, s := range d.sessions { | ||
| 219 | snapshot = append(snapshot, s) | 309 | snapshot = append(snapshot, s) | ||
| @@ -349,6 +439,14 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||||
| 349 | } | 439 | } | ||
| 350 | 440 | ||||
| 351 | d.mu.Lock() | 441 | d.mu.Lock() | ||
| 442 | + | // Shutdown/create interlock: once shutdown has begun (last session removed, | |||
| 443 | + | // or shutdownAll started) no new session may be created and no attach may | |||
| 444 | + | // proceed — the teardown snapshot has already been (or is being) taken, so | |||
| 445 | + | // anything inserted now would leak its PTY and hang Serve. | |||
| 446 | + | if d.closing { | |||
| 447 | + | d.mu.Unlock() | |||
| 448 | + | return nil, nil, &protoErr{ports.ErrServerShutdown, "daemon is shutting down"} | |||
| 449 | + | } | |||
| 352 | switch h.Intent { | 450 | switch h.Intent { | ||
| 353 | case ports.IntentEphemeral: | 451 | case ports.IntentEphemeral: | ||
| 354 | name := d.allocEphemeralNameLocked() | 452 | name := d.allocEphemeralNameLocked() | ||
| @@ -438,8 +536,9 @@ func (d *Daemon) attachClient(sess *session, tr ports.Transport, sz domain.Size) | |||||
| 438 | sess.mu.Unlock() | 536 | sess.mu.Unlock() | ||
| 439 | 537 | ||||
| 440 | if old != nil { | 538 | if old != nil { | ||
| 441 | - | _ = old.send(frameDetached(ports.ReasonDetach)) | 539 | + | // Async + bounded: a dead or wedged old client must not stall the new |
| 442 | - | _ = old.tr.Close() | 540 | + | // client's handshake. |
| 541 | + | d.notifyDetachedAsync(old, ports.ReasonDetach) | |||
| 443 | } | 542 | } | ||
| 444 | return ac | 543 | return ac | ||
| 445 | } | 544 | } | ||
| @@ -499,7 +598,11 @@ func (d *Daemon) clientGone(sess *session, ac *attachedClient, explicit bool) { | |||||
| 499 | return // already displaced by a newer client; nothing to do | 598 | return // already displaced by a newer client; nothing to do | ||
| 500 | } | 599 | } | ||
| 501 | if explicit { | 600 | if explicit { | ||
| 502 | - | _ = ac.send(frameDetached(ports.ReasonDetach)) | 601 | + | // Synchronous so the ack is delivered before the transport closes |
| 602 | + | // (the client is actively awaiting it), but deadline-bounded so a | |||
| 603 | + | // wedged client cannot pin this conn handler and hang Serve's | |||
| 604 | + | // connWg.Wait. | |||
| 605 | + | d.boundedSend(ac, frameDetached(ports.ReasonDetach)) | |||
| 503 | } | 606 | } | ||
| 504 | _ = ac.tr.Close() | 607 | _ = ac.tr.Close() | ||
| 505 | if sess.ephemeral { | 608 | if sess.ephemeral { | ||
| @@ -657,18 +760,27 @@ func (d *Daemon) resize(sess *session, ac *attachedClient, sz domain.Size) { | |||||
| 657 | } | 760 | } | ||
| 658 | } | 761 | } | ||
| 659 | 762 | ||||
| 660 | - | // detachOnSendError drops a client whose transport failed but keeps the session | 763 | + | // detachOnSendError drops a client whose transport failed. Like every other |
| 661 | - | // running headless (per plan: output errors never kill the session). | 764 | + | // detach path, losing the client kills an ephemeral session (its lifetime is |
| 765 | + | // its client's); a named session keeps running headless. | |||
| 662 | func (d *Daemon) detachOnSendError(sess *session, ac *attachedClient) { | 766 | func (d *Daemon) detachOnSendError(sess *session, ac *attachedClient) { | ||
| 663 | if sess.detachIfCurrent(ac) { | 767 | if sess.detachIfCurrent(ac) { | ||
| 664 | _ = ac.tr.Close() | 768 | _ = ac.tr.Close() | ||
| 665 | d.log.Warn("detached client after send error", "session", sess.name) | 769 | d.log.Warn("detached client after send error", "session", sess.name) | ||
| 770 | + | if sess.ephemeral { | |||
| 771 | + | d.killSession(sess, ports.ReasonSessionKilled) | |||
| 772 | + | } | |||
| 666 | } | 773 | } | ||
| 667 | } | 774 | } | ||
| 668 | 775 | ||||
| 669 | // killSession removes a session and tears down its resources. It is | 776 | // killSession removes a session and tears down its resources. It is | ||
| 670 | // idempotent: only the caller that wins the registry delete acts. When the | 777 | // idempotent: only the caller that wins the registry delete acts. When the | ||
| 671 | - | // registry empties it signals daemon shutdown. | 778 | + | // registry empties it marks the daemon closing (atomically with the |
| 779 | + | // empty-check, under d.mu) and signals shutdown. | |||
| 780 | + | // | |||
| 781 | + | // Teardown ordering matters: context cancel, pty.Close, and the done signal | |||
| 782 | + | // run first and unconditionally — never gated behind a client send. The | |||
| 783 | + | // Detached notice is best-effort, asynchronous, and deadline-bounded. | |||
| 672 | func (d *Daemon) killSession(sess *session, reason uint8) { | 784 | func (d *Daemon) killSession(sess *session, reason uint8) { | ||
| 673 | d.mu.Lock() | 785 | d.mu.Lock() | ||
| 674 | if _, ok := d.sessions[sess.id]; !ok { | 786 | if _, ok := d.sessions[sess.id]; !ok { | ||
| @@ -677,23 +789,28 @@ func (d *Daemon) killSession(sess *session, reason uint8) { | |||||
| 677 | } | 789 | } | ||
| 678 | delete(d.sessions, sess.id) | 790 | delete(d.sessions, sess.id) | ||
| 679 | empty := len(d.sessions) == 0 | 791 | empty := len(d.sessions) == 0 | ||
| 792 | + | if empty { | |||
| 793 | + | // Shutdown is now inevitable and irreversible (doneOnce below): stop | |||
| 794 | + | // route from inserting new sessions from this instant, while we still | |||
| 795 | + | // hold the same lock route checks under. | |||
| 796 | + | d.closing = true | |||
| 797 | + | } | |||
| 680 | d.mu.Unlock() | 798 | d.mu.Unlock() | ||
| 681 | 799 | ||||
| 682 | sess.mu.Lock() | 800 | sess.mu.Lock() | ||
| 683 | ac := sess.client | 801 | ac := sess.client | ||
| 684 | sess.client = nil | 802 | sess.client = nil | ||
| 685 | sess.mu.Unlock() | 803 | sess.mu.Unlock() | ||
| 686 | - | if ac != nil { | |||
| 687 | - | _ = ac.send(frameDetached(reason)) | |||
| 688 | - | _ = ac.tr.Close() | |||
| 689 | - | } | |||
| 690 | 804 | ||||
| 691 | sess.cancel() | 805 | sess.cancel() | ||
| 692 | _ = sess.win.pty.Close() | 806 | _ = sess.win.pty.Close() | ||
| 693 | - | ||||
| 694 | if empty { | 807 | if empty { | ||
| 695 | d.doneOnce.Do(func() { close(d.done) }) | 808 | d.doneOnce.Do(func() { close(d.done) }) | ||
| 696 | } | 809 | } | ||
| 810 | + | ||||
| 811 | + | if ac != nil { | |||
| 812 | + | d.notifyDetachedAsync(ac, reason) | |||
| 813 | + | } | |||
| 697 | } | 814 | } | ||
| 698 | 815 | ||||
| 699 | // allocEphemeralNameLocked returns the lowest free decimal name. Caller holds | 816 | // allocEphemeralNameLocked returns the lowest free decimal name. Caller holds | ||
M internal/usecase/daemon/daemon_test.go +185 -0
| @@ -190,6 +190,7 @@ func TestHandshakeNewHappy(t *testing.T) { | |||
| 190 | 190 | releasePTY() | |
| 191 | 191 | hg.Wait() | |
| 192 | 192 | d.sessWg.Wait() | |
| 193 | + | d.waitNotifies() | |
| 193 | 194 | } | |
| 194 | 195 | ||
| 195 | 196 | func TestHandshakeVersionMismatch(t *testing.T) { | |
| @@ -284,6 +285,7 @@ func TestAttachReplaceDetachesOld(t *testing.T) { | |||
| 284 | 285 | releasePTY() | |
| 285 | 286 | hg.Wait() | |
| 286 | 287 | d.sessWg.Wait() | |
| 288 | + | d.waitNotifies() | |
| 287 | 289 | } | |
| 288 | 290 | ||
| 289 | 291 | // --- detach semantics ------------------------------------------------------- | |
| @@ -333,6 +335,7 @@ func TestDetachKeepsNamed(t *testing.T) { | |||
| 333 | 335 | d.killSession(sess, ports.ReasonServerShutdown) | |
| 334 | 336 | releasePTY() | |
| 335 | 337 | d.sessWg.Wait() | |
| 338 | + | d.waitNotifies() | |
| 336 | 339 | } | |
| 337 | 340 | ||
| 338 | 341 | // --- render scheduler debounce ---------------------------------------------- | |
| @@ -466,6 +469,7 @@ func TestReaderEOFRemovesSessionAndSignalsShutdown(t *testing.T) { | |||
| 466 | 469 | ||
| 467 | 470 | hg.Wait() | |
| 468 | 471 | d.sessWg.Wait() | |
| 472 | + | d.waitNotifies() | |
| 469 | 473 | } | |
| 470 | 474 | ||
| 471 | 475 | // --- full Serve lifecycle --------------------------------------------------- | |
| @@ -552,3 +556,184 @@ func TestServeGracefulShutdownOnContextCancel(t *testing.T) { | |||
| 552 | 556 | t.Fatal("Serve did not return after context cancel") | |
| 553 | 557 | } | |
| 554 | 558 | } | |
| 559 | + | ||
| 560 | + | // --- shutdown/create interlock ------------------------------------------------ | |
| 561 | + | ||
| 562 | + | // TestHelloRacingShutdownIsRejected covers the interlock between the | |
| 563 | + | // registry-empty shutdown decision and session creation: once killSession has | |
| 564 | + | // removed the last session (making shutdown irreversible via doneOnce), a | |
| 565 | + | // Hello landing at any later instant must be rejected cleanly — never spawn a | |
| 566 | + | // child that no teardown pass would reap. The interleaving is deterministic: | |
| 567 | + | // killSession runs to completion before the racing Hellos are handled, which | |
| 568 | + | // models the worst case (insertion attempt after the shutdown snapshot). | |
| 569 | + | func TestHelloRacingShutdownIsRejected(t *testing.T) { | |
| 570 | + | p, releasePTY := newBlockingPTY(t) | |
| 571 | + | f := portsmocks.NewMockPTYFactory(t) | |
| 572 | + | // Exactly one child may ever be spawned; a second Open call (a leaked | |
| 573 | + | // child) fails the test via mock expectations. | |
| 574 | + | f.EXPECT().Open(mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(p, nil).Once() | |
| 575 | + | d := newTestDaemon(t, f, stubClock{}) | |
| 576 | + | ||
| 577 | + | tr1, sends1, release1 := newConn(t, mustHello(ports.IntentEphemeral, "", domain.Size{Cols: 80, Rows: 24})) | |
| 578 | + | var hg sync.WaitGroup | |
| 579 | + | hg.Add(1) | |
| 580 | + | go func() { defer hg.Done(); d.handleConn(tr1) }() | |
| 581 | + | awaitFrame(t, sends1, ports.MsgWelcome) | |
| 582 | + | sess := firstSession(d) | |
| 583 | + | require.NotNil(t, sess) | |
| 584 | + | ||
| 585 | + | // The last session dies: shutdown begins irreversibly. | |
| 586 | + | d.killSession(sess, ports.ReasonSessionKilled) | |
| 587 | + | select { | |
| 588 | + | case <-d.done: | |
| 589 | + | default: | |
| 590 | + | t.Fatal("registry-empty shutdown must be signalled synchronously by killSession") | |
| 591 | + | } | |
| 592 | + | ||
| 593 | + | // Racing Hellos — both creation intents must be rejected with a clean | |
| 594 | + | // typed error, and nothing may be inserted into the registry. | |
| 595 | + | for _, intent := range []struct { | |
| 596 | + | name string | |
| 597 | + | intent uint8 | |
| 598 | + | sess string | |
| 599 | + | }{ | |
| 600 | + | {"ephemeral", ports.IntentEphemeral, ""}, | |
| 601 | + | {"named", ports.IntentNew, "latecomer"}, | |
| 602 | + | } { | |
| 603 | + | tr2, sends2, _ := newConn(t, mustHello(intent.intent, intent.sess, domain.Size{Cols: 80, Rows: 24})) | |
| 604 | + | d.handleConn(tr2) | |
| 605 | + | e := awaitFrame(t, sends2, ports.MsgError) | |
| 606 | + | em, err := ports.UnmarshalErrorMsg(e.Payload) | |
| 607 | + | require.NoError(t, err) | |
| 608 | + | require.Equal(t, ports.ErrServerShutdown, em.Code, "%s hello racing shutdown must be rejected", intent.name) | |
| 609 | + | } | |
| 610 | + | require.Equal(t, 0, sessionCount(d), "no session may be inserted after shutdown began") | |
| 611 | + | ||
| 612 | + | release1() | |
| 613 | + | releasePTY() | |
| 614 | + | hg.Wait() | |
| 615 | + | d.sessWg.Wait() | |
| 616 | + | d.waitNotifies() | |
| 617 | + | } | |
| 618 | + | ||
| 619 | + | // --- wedged-client teardown --------------------------------------------------- | |
| 620 | + | ||
| 621 | + | // notifyClock drives the daemon with a stub debounce timer (schedulers park) | |
| 622 | + | // but a short real timer for the detach-notify deadline, so a wedged client's | |
| 623 | + | // transport is force-closed quickly and deterministically. | |
| 624 | + | type notifyClock struct{} | |
| 625 | + | ||
| 626 | + | func (notifyClock) Now() time.Time { return time.Now() } | |
| 627 | + | func (notifyClock) NewTimer(d time.Duration) ports.Timer { | |
| 628 | + | if d == debounceInterval { | |
| 629 | + | return stubTimer{} | |
| 630 | + | } | |
| 631 | + | return realTimer{t: time.NewTimer(5 * time.Millisecond)} | |
| 632 | + | } | |
| 633 | + | ||
| 634 | + | type realTimer struct{ t *time.Timer } | |
| 635 | + | ||
| 636 | + | func (r realTimer) C() <-chan time.Time { return r.t.C } | |
| 637 | + | func (r realTimer) Reset(d time.Duration) bool { return r.t.Reset(d) } | |
| 638 | + | func (r realTimer) Stop() bool { return r.t.Stop() } | |
| 639 | + | ||
| 640 | + | // TestServeReturnsDespiteWedgedClientOnShutdown: killSession's teardown must | |
| 641 | + | // not be gated behind the Detached notice. The client transport wedges every | |
| 642 | + | // MsgDetached send until the transport is closed (mirroring a full kernel | |
| 643 | + | // send buffer: only Close fails the in-flight write). Serve must still tear | |
| 644 | + | // down the PTY and return. | |
| 645 | + | func TestServeReturnsDespiteWedgedClientOnShutdown(t *testing.T) { | |
| 646 | + | p, _ := newBlockingPTY(t) // Close unblocks the parked reader | |
| 647 | + | ||
| 648 | + | tr := portsmocks.NewMockTransport(t) | |
| 649 | + | sends := make(chan ports.Frame, 64) | |
| 650 | + | recvCh := make(chan ports.Frame, 1) | |
| 651 | + | recvCh <- mustHello(ports.IntentNew, "wedge", domain.Size{Cols: 80, Rows: 24}) | |
| 652 | + | connDone := make(chan struct{}) | |
| 653 | + | var connOnce sync.Once | |
| 654 | + | closeConn := func() { connOnce.Do(func() { close(connDone) }) } | |
| 655 | + | ||
| 656 | + | tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) { | |
| 657 | + | select { | |
| 658 | + | case f := <-recvCh: | |
| 659 | + | return f, nil | |
| 660 | + | case <-connDone: | |
| 661 | + | return ports.Frame{}, io.EOF | |
| 662 | + | } | |
| 663 | + | }).Maybe() | |
| 664 | + | tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f ports.Frame) error { | |
| 665 | + | if f.Type == ports.MsgDetached { | |
| 666 | + | // Wedged: blocks until the transport is force-closed. | |
| 667 | + | <-connDone | |
| 668 | + | return io.ErrClosedPipe | |
| 669 | + | } | |
| 670 | + | sends <- f | |
| 671 | + | return nil | |
| 672 | + | }).Maybe() | |
| 673 | + | tr.EXPECT().Close().RunAndReturn(func() error { closeConn(); return nil }).Maybe() | |
| 674 | + | ||
| 675 | + | l := portsmocks.NewMockListener(t) | |
| 676 | + | connCh := make(chan ports.Transport, 1) | |
| 677 | + | connCh <- tr | |
| 678 | + | lnClosed := make(chan struct{}) | |
| 679 | + | var lnOnce sync.Once | |
| 680 | + | l.EXPECT().Accept().RunAndReturn(func() (ports.Transport, error) { | |
| 681 | + | select { | |
| 682 | + | case c := <-connCh: | |
| 683 | + | return c, nil | |
| 684 | + | case <-lnClosed: | |
| 685 | + | return nil, io.EOF | |
| 686 | + | } | |
| 687 | + | }).Maybe() | |
| 688 | + | l.EXPECT().Close().RunAndReturn(func() error { lnOnce.Do(func() { close(lnClosed) }); return nil }).Maybe() | |
| 689 | + | l.EXPECT().Addr().Return("mock").Maybe() | |
| 690 | + | ||
| 691 | + | d := New(newFactory(t, p), notifyClock{}, slog.New(slog.NewTextHandler(io.Discard, nil))) | |
| 692 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 693 | + | ||
| 694 | + | served := make(chan error, 1) | |
| 695 | + | go func() { served <- d.Serve(ctx, l) }() | |
| 696 | + | ||
| 697 | + | awaitFrame(t, sends, ports.MsgWelcome) | |
| 698 | + | awaitFrame(t, sends, ports.MsgOutput) | |
| 699 | + | ||
| 700 | + | cancel() // shutdown with the wedged client still attached | |
| 701 | + | ||
| 702 | + | select { | |
| 703 | + | case err := <-served: | |
| 704 | + | require.NoError(t, err) | |
| 705 | + | case <-time.After(3 * time.Second): | |
| 706 | + | t.Fatal("Serve hung on a wedged client during teardown") | |
| 707 | + | } | |
| 708 | + | require.Equal(t, 0, sessionCount(d), "session must be torn down despite the wedged client") | |
| 709 | + | } | |
| 710 | + | ||
| 711 | + | // --- send-error kills ephemeral ----------------------------------------------- | |
| 712 | + | ||
| 713 | + | // TestSendErrorKillsEphemeral: a failed output send detaches the client, and — | |
| 714 | + | // like every other detach path — that kills an ephemeral session rather than | |
| 715 | + | // leaving it headless-but-alive forever. | |
| 716 | + | func TestSendErrorKillsEphemeral(t *testing.T) { | |
| 717 | + | p := portsmocks.NewMockPTY(t) | |
| 718 | + | p.EXPECT().Close().Return(nil).Maybe() | |
| 719 | + | ||
| 720 | + | tr := portsmocks.NewMockTransport(t) | |
| 721 | + | tr.EXPECT().Send(mock.Anything).Return(io.ErrClosedPipe).Maybe() | |
| 722 | + | tr.EXPECT().Close().Return(nil).Maybe() | |
| 723 | + | ||
| 724 | + | d := newTestDaemon(t, nil, stubClock{}) | |
| 725 | + | win := &window{pty: p, screen: vt.NewScreen(20, 5), dirty: make(chan struct{}, 1), size: domain.Size{Cols: 20, Rows: 5}} | |
| 726 | + | sctx, cancel := context.WithCancel(context.Background()) | |
| 727 | + | ac := &attachedClient{tr: tr, rend: renderer.New(renderer.Capabilities{})} | |
| 728 | + | sess := &session{id: "e", name: "0", ephemeral: true, win: win, ctx: sctx, cancel: cancel, client: ac} | |
| 729 | + | d.sessions[sess.id] = sess | |
| 730 | + | ||
| 731 | + | win.mu.Lock() | |
| 732 | + | win.screen.Write([]byte("x")) | |
| 733 | + | win.mu.Unlock() | |
| 734 | + | ||
| 735 | + | d.paint(sess, ac, true) // send fails -> detach -> ephemeral killed | |
| 736 | + | ||
| 737 | + | require.Equal(t, 0, sessionCount(d), "ephemeral session must die when its client's send fails") | |
| 738 | + | d.waitNotifies() | |
| 739 | + | } | |
| @@ -190,6 +190,7 @@ func TestHandshakeNewHappy(t *testing.T) { | |||||
| 190 | releasePTY() | 190 | releasePTY() | ||
| 191 | hg.Wait() | 191 | hg.Wait() | ||
| 192 | d.sessWg.Wait() | 192 | d.sessWg.Wait() | ||
| 193 | + | d.waitNotifies() | |||
| 193 | } | 194 | } | ||
| 194 | 195 | ||||
| 195 | func TestHandshakeVersionMismatch(t *testing.T) { | 196 | func TestHandshakeVersionMismatch(t *testing.T) { | ||
| @@ -284,6 +285,7 @@ func TestAttachReplaceDetachesOld(t *testing.T) { | |||||
| 284 | releasePTY() | 285 | releasePTY() | ||
| 285 | hg.Wait() | 286 | hg.Wait() | ||
| 286 | d.sessWg.Wait() | 287 | d.sessWg.Wait() | ||
| 288 | + | d.waitNotifies() | |||
| 287 | } | 289 | } | ||
| 288 | 290 | ||||
| 289 | // --- detach semantics ------------------------------------------------------- | 291 | // --- detach semantics ------------------------------------------------------- | ||
| @@ -333,6 +335,7 @@ func TestDetachKeepsNamed(t *testing.T) { | |||||
| 333 | d.killSession(sess, ports.ReasonServerShutdown) | 335 | d.killSession(sess, ports.ReasonServerShutdown) | ||
| 334 | releasePTY() | 336 | releasePTY() | ||
| 335 | d.sessWg.Wait() | 337 | d.sessWg.Wait() | ||
| 338 | + | d.waitNotifies() | |||
| 336 | } | 339 | } | ||
| 337 | 340 | ||||
| 338 | // --- render scheduler debounce ---------------------------------------------- | 341 | // --- render scheduler debounce ---------------------------------------------- | ||
| @@ -466,6 +469,7 @@ func TestReaderEOFRemovesSessionAndSignalsShutdown(t *testing.T) { | |||||
| 466 | 469 | ||||
| 467 | hg.Wait() | 470 | hg.Wait() | ||
| 468 | d.sessWg.Wait() | 471 | d.sessWg.Wait() | ||
| 472 | + | d.waitNotifies() | |||
| 469 | } | 473 | } | ||
| 470 | 474 | ||||
| 471 | // --- full Serve lifecycle --------------------------------------------------- | 475 | // --- full Serve lifecycle --------------------------------------------------- | ||
| @@ -552,3 +556,184 @@ func TestServeGracefulShutdownOnContextCancel(t *testing.T) { | |||||
| 552 | t.Fatal("Serve did not return after context cancel") | 556 | t.Fatal("Serve did not return after context cancel") | ||
| 553 | } | 557 | } | ||
| 554 | } | 558 | } | ||
| 559 | + | ||||
| 560 | + | // --- shutdown/create interlock ------------------------------------------------ | |||
| 561 | + | ||||
| 562 | + | // TestHelloRacingShutdownIsRejected covers the interlock between the | |||
| 563 | + | // registry-empty shutdown decision and session creation: once killSession has | |||
| 564 | + | // removed the last session (making shutdown irreversible via doneOnce), a | |||
| 565 | + | // Hello landing at any later instant must be rejected cleanly — never spawn a | |||
| 566 | + | // child that no teardown pass would reap. The interleaving is deterministic: | |||
| 567 | + | // killSession runs to completion before the racing Hellos are handled, which | |||
| 568 | + | // models the worst case (insertion attempt after the shutdown snapshot). | |||
| 569 | + | func TestHelloRacingShutdownIsRejected(t *testing.T) { | |||
| 570 | + | p, releasePTY := newBlockingPTY(t) | |||
| 571 | + | f := portsmocks.NewMockPTYFactory(t) | |||
| 572 | + | // Exactly one child may ever be spawned; a second Open call (a leaked | |||
| 573 | + | // child) fails the test via mock expectations. | |||
| 574 | + | f.EXPECT().Open(mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(p, nil).Once() | |||
| 575 | + | d := newTestDaemon(t, f, stubClock{}) | |||
| 576 | + | ||||
| 577 | + | tr1, sends1, release1 := newConn(t, mustHello(ports.IntentEphemeral, "", domain.Size{Cols: 80, Rows: 24})) | |||
| 578 | + | var hg sync.WaitGroup | |||
| 579 | + | hg.Add(1) | |||
| 580 | + | go func() { defer hg.Done(); d.handleConn(tr1) }() | |||
| 581 | + | awaitFrame(t, sends1, ports.MsgWelcome) | |||
| 582 | + | sess := firstSession(d) | |||
| 583 | + | require.NotNil(t, sess) | |||
| 584 | + | ||||
| 585 | + | // The last session dies: shutdown begins irreversibly. | |||
| 586 | + | d.killSession(sess, ports.ReasonSessionKilled) | |||
| 587 | + | select { | |||
| 588 | + | case <-d.done: | |||
| 589 | + | default: | |||
| 590 | + | t.Fatal("registry-empty shutdown must be signalled synchronously by killSession") | |||
| 591 | + | } | |||
| 592 | + | ||||
| 593 | + | // Racing Hellos — both creation intents must be rejected with a clean | |||
| 594 | + | // typed error, and nothing may be inserted into the registry. | |||
| 595 | + | for _, intent := range []struct { | |||
| 596 | + | name string | |||
| 597 | + | intent uint8 | |||
| 598 | + | sess string | |||
| 599 | + | }{ | |||
| 600 | + | {"ephemeral", ports.IntentEphemeral, ""}, | |||
| 601 | + | {"named", ports.IntentNew, "latecomer"}, | |||
| 602 | + | } { | |||
| 603 | + | tr2, sends2, _ := newConn(t, mustHello(intent.intent, intent.sess, domain.Size{Cols: 80, Rows: 24})) | |||
| 604 | + | d.handleConn(tr2) | |||
| 605 | + | e := awaitFrame(t, sends2, ports.MsgError) | |||
| 606 | + | em, err := ports.UnmarshalErrorMsg(e.Payload) | |||
| 607 | + | require.NoError(t, err) | |||
| 608 | + | require.Equal(t, ports.ErrServerShutdown, em.Code, "%s hello racing shutdown must be rejected", intent.name) | |||
| 609 | + | } | |||
| 610 | + | require.Equal(t, 0, sessionCount(d), "no session may be inserted after shutdown began") | |||
| 611 | + | ||||
| 612 | + | release1() | |||
| 613 | + | releasePTY() | |||
| 614 | + | hg.Wait() | |||
| 615 | + | d.sessWg.Wait() | |||
| 616 | + | d.waitNotifies() | |||
| 617 | + | } | |||
| 618 | + | ||||
| 619 | + | // --- wedged-client teardown --------------------------------------------------- | |||
| 620 | + | ||||
| 621 | + | // notifyClock drives the daemon with a stub debounce timer (schedulers park) | |||
| 622 | + | // but a short real timer for the detach-notify deadline, so a wedged client's | |||
| 623 | + | // transport is force-closed quickly and deterministically. | |||
| 624 | + | type notifyClock struct{} | |||
| 625 | + | ||||
| 626 | + | func (notifyClock) Now() time.Time { return time.Now() } | |||
| 627 | + | func (notifyClock) NewTimer(d time.Duration) ports.Timer { | |||
| 628 | + | if d == debounceInterval { | |||
| 629 | + | return stubTimer{} | |||
| 630 | + | } | |||
| 631 | + | return realTimer{t: time.NewTimer(5 * time.Millisecond)} | |||
| 632 | + | } | |||
| 633 | + | ||||
| 634 | + | type realTimer struct{ t *time.Timer } | |||
| 635 | + | ||||
| 636 | + | func (r realTimer) C() <-chan time.Time { return r.t.C } | |||
| 637 | + | func (r realTimer) Reset(d time.Duration) bool { return r.t.Reset(d) } | |||
| 638 | + | func (r realTimer) Stop() bool { return r.t.Stop() } | |||
| 639 | + | ||||
| 640 | + | // TestServeReturnsDespiteWedgedClientOnShutdown: killSession's teardown must | |||
| 641 | + | // not be gated behind the Detached notice. The client transport wedges every | |||
| 642 | + | // MsgDetached send until the transport is closed (mirroring a full kernel | |||
| 643 | + | // send buffer: only Close fails the in-flight write). Serve must still tear | |||
| 644 | + | // down the PTY and return. | |||
| 645 | + | func TestServeReturnsDespiteWedgedClientOnShutdown(t *testing.T) { | |||
| 646 | + | p, _ := newBlockingPTY(t) // Close unblocks the parked reader | |||
| 647 | + | ||||
| 648 | + | tr := portsmocks.NewMockTransport(t) | |||
| 649 | + | sends := make(chan ports.Frame, 64) | |||
| 650 | + | recvCh := make(chan ports.Frame, 1) | |||
| 651 | + | recvCh <- mustHello(ports.IntentNew, "wedge", domain.Size{Cols: 80, Rows: 24}) | |||
| 652 | + | connDone := make(chan struct{}) | |||
| 653 | + | var connOnce sync.Once | |||
| 654 | + | closeConn := func() { connOnce.Do(func() { close(connDone) }) } | |||
| 655 | + | ||||
| 656 | + | tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) { | |||
| 657 | + | select { | |||
| 658 | + | case f := <-recvCh: | |||
| 659 | + | return f, nil | |||
| 660 | + | case <-connDone: | |||
| 661 | + | return ports.Frame{}, io.EOF | |||
| 662 | + | } | |||
| 663 | + | }).Maybe() | |||
| 664 | + | tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f ports.Frame) error { | |||
| 665 | + | if f.Type == ports.MsgDetached { | |||
| 666 | + | // Wedged: blocks until the transport is force-closed. | |||
| 667 | + | <-connDone | |||
| 668 | + | return io.ErrClosedPipe | |||
| 669 | + | } | |||
| 670 | + | sends <- f | |||
| 671 | + | return nil | |||
| 672 | + | }).Maybe() | |||
| 673 | + | tr.EXPECT().Close().RunAndReturn(func() error { closeConn(); return nil }).Maybe() | |||
| 674 | + | ||||
| 675 | + | l := portsmocks.NewMockListener(t) | |||
| 676 | + | connCh := make(chan ports.Transport, 1) | |||
| 677 | + | connCh <- tr | |||
| 678 | + | lnClosed := make(chan struct{}) | |||
| 679 | + | var lnOnce sync.Once | |||
| 680 | + | l.EXPECT().Accept().RunAndReturn(func() (ports.Transport, error) { | |||
| 681 | + | select { | |||
| 682 | + | case c := <-connCh: | |||
| 683 | + | return c, nil | |||
| 684 | + | case <-lnClosed: | |||
| 685 | + | return nil, io.EOF | |||
| 686 | + | } | |||
| 687 | + | }).Maybe() | |||
| 688 | + | l.EXPECT().Close().RunAndReturn(func() error { lnOnce.Do(func() { close(lnClosed) }); return nil }).Maybe() | |||
| 689 | + | l.EXPECT().Addr().Return("mock").Maybe() | |||
| 690 | + | ||||
| 691 | + | d := New(newFactory(t, p), notifyClock{}, slog.New(slog.NewTextHandler(io.Discard, nil))) | |||
| 692 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 693 | + | ||||
| 694 | + | served := make(chan error, 1) | |||
| 695 | + | go func() { served <- d.Serve(ctx, l) }() | |||
| 696 | + | ||||
| 697 | + | awaitFrame(t, sends, ports.MsgWelcome) | |||
| 698 | + | awaitFrame(t, sends, ports.MsgOutput) | |||
| 699 | + | ||||
| 700 | + | cancel() // shutdown with the wedged client still attached | |||
| 701 | + | ||||
| 702 | + | select { | |||
| 703 | + | case err := <-served: | |||
| 704 | + | require.NoError(t, err) | |||
| 705 | + | case <-time.After(3 * time.Second): | |||
| 706 | + | t.Fatal("Serve hung on a wedged client during teardown") | |||
| 707 | + | } | |||
| 708 | + | require.Equal(t, 0, sessionCount(d), "session must be torn down despite the wedged client") | |||
| 709 | + | } | |||
| 710 | + | ||||
| 711 | + | // --- send-error kills ephemeral ----------------------------------------------- | |||
| 712 | + | ||||
| 713 | + | // TestSendErrorKillsEphemeral: a failed output send detaches the client, and — | |||
| 714 | + | // like every other detach path — that kills an ephemeral session rather than | |||
| 715 | + | // leaving it headless-but-alive forever. | |||
| 716 | + | func TestSendErrorKillsEphemeral(t *testing.T) { | |||
| 717 | + | p := portsmocks.NewMockPTY(t) | |||
| 718 | + | p.EXPECT().Close().Return(nil).Maybe() | |||
| 719 | + | ||||
| 720 | + | tr := portsmocks.NewMockTransport(t) | |||
| 721 | + | tr.EXPECT().Send(mock.Anything).Return(io.ErrClosedPipe).Maybe() | |||
| 722 | + | tr.EXPECT().Close().Return(nil).Maybe() | |||
| 723 | + | ||||
| 724 | + | d := newTestDaemon(t, nil, stubClock{}) | |||
| 725 | + | win := &window{pty: p, screen: vt.NewScreen(20, 5), dirty: make(chan struct{}, 1), size: domain.Size{Cols: 20, Rows: 5}} | |||
| 726 | + | sctx, cancel := context.WithCancel(context.Background()) | |||
| 727 | + | ac := &attachedClient{tr: tr, rend: renderer.New(renderer.Capabilities{})} | |||
| 728 | + | sess := &session{id: "e", name: "0", ephemeral: true, win: win, ctx: sctx, cancel: cancel, client: ac} | |||
| 729 | + | d.sessions[sess.id] = sess | |||
| 730 | + | ||||
| 731 | + | win.mu.Lock() | |||
| 732 | + | win.screen.Write([]byte("x")) | |||
| 733 | + | win.mu.Unlock() | |||
| 734 | + | ||||
| 735 | + | d.paint(sess, ac, true) // send fails -> detach -> ephemeral killed | |||
| 736 | + | ||||
| 737 | + | require.Equal(t, 0, sessionCount(d), "ephemeral session must die when its client's send fails") | |||
| 738 | + | d.waitNotifies() | |||
| 739 | + | } | |||