detached ee9b61d

fix(phase5): bound final cancellation and readiness races

commit
ee9b61ddfaaa6a7b5a0f0d453dadc5553c15ca8b
author
brice · · 2026-08-03 17:18 UTC
committer
brice · · 2026-08-04 13:34 UTC
parent
448248d
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
9 files · +387 -30
Split
UnifiedSplit
M internal/adapters/sshstdio/transport.go +181 -10
@@ -12,6 +12,7 @@ import (
1212 "fmt"
1313 "io"
1414 "log/slog"
15+ "os"
1516 "os/exec"
1617 "strings"
1718 "sync"
@@ -69,7 +70,26 @@ func newTransport(r io.Reader, w io.Writer, closeFn closeFunc, eofErr eofErrFunc
6970 if closeFn == nil {
7071 closeFn = func() error { return nil }
7172 }
72- return &transport{r: r, w: w, close: closeFn, eofErr: eofErr}
73+ t := &transport{
74+ r: r,
75+ w: w,
76+ close: closeFn,
77+ eofErr: eofErr,
78+ done: make(chan struct{}),
79+ }
80+ if file, ok := r.(*os.File); ok && file == os.Stdin {
81+ t.r = newProcessStdinReader(file, t.done)
82+ } else if closer, ok := r.(io.Closer); ok {
83+ t.readerCloser = closer
84+ } else {
85+ t.r = newUnownedReader(r, t.done)
86+ }
87+ if file, ok := w.(*os.File); !ok || file != os.Stdout {
88+ if closer, ok := w.(io.Closer); ok {
89+ t.writerCloser = closer
90+ }
91+ }
92+ return t
7393 }
7494
7595 type transport struct {
@@ -77,6 +97,9 @@ type transport struct {
7797 w io.Writer
7898 close closeFunc
7999 eofErr eofErrFunc
100+ readerCloser io.Closer
101+ writerCloser io.Closer
102+ done chan struct{}
80103 observer ports.SerializedRuntimeObserver
81104 operationMu sync.Mutex
82105 operationCount [operationKindCount]int
@@ -89,6 +112,154 @@ type transport struct {
89112 readBuf []byte
90113 }
91114
115+type readResult struct {
116+ data []byte
117+ err error
118+}
119+
120+// unownedReader is the generic cancellation boundary for readers that cannot
121+// be closed by the transport. The worker may remain blocked until the source
122+// itself is released; process stdin uses the singleton pump below instead.
123+// ponytail: arbitrary io.Reader has no cancellation contract, so one fallback
124+// worker may remain until that source returns; add a cancellable reader port if
125+// another production unowned source appears.
126+type unownedReader struct {
127+ r io.Reader
128+ done <-chan struct{}
129+}
130+
131+func newUnownedReader(r io.Reader, done <-chan struct{}) *unownedReader {
132+ return &unownedReader{r: r, done: done}
133+}
134+
135+func (r *unownedReader) Read(dst []byte) (int, error) {
136+ select {
137+ case <-r.done:
138+ return 0, io.ErrClosedPipe
139+ default:
140+ }
141+ result := make(chan readResult, 1)
142+ go func() {
143+ buf := make([]byte, len(dst))
144+ n, err := r.r.Read(buf)
145+ result <- readResult{data: buf[:n], err: err}
146+ }()
147+ select {
148+ case <-r.done:
149+ return 0, io.ErrClosedPipe
150+ case result := <-result:
151+ select {
152+ case <-r.done:
153+ return 0, io.ErrClosedPipe
154+ default:
155+ }
156+ copy(dst, result.data)
157+ return len(result.data), result.err
158+ }
159+}
160+
161+type processStdinPump struct {
162+ chunks chan readResult
163+ mu sync.Mutex
164+ buf []byte
165+ err error
166+}
167+
168+var processStdinPumps struct {
169+ sync.Mutex
170+ file *os.File
171+ pump *processStdinPump
172+}
173+
174+// processStdinPumpFor keeps one read worker for the process stdin lifetime.
175+// Reconnects therefore do not strand one blocked worker per transport.
176+func processStdinPumpFor(file *os.File) *processStdinPump {
177+ processStdinPumps.Lock()
178+ defer processStdinPumps.Unlock()
179+ if processStdinPumps.pump != nil && processStdinPumps.file == file {
180+ return processStdinPumps.pump
181+ }
182+ pump := &processStdinPump{chunks: make(chan readResult, 1)}
183+ processStdinPumps.file = file
184+ processStdinPumps.pump = pump
185+ go pump.run(file)
186+ return pump
187+}
188+
189+func (p *processStdinPump) run(file *os.File) {
190+ for {
191+ buf := make([]byte, 32*1024)
192+ n, err := file.Read(buf)
193+ if n > 0 {
194+ p.chunks <- readResult{data: append([]byte(nil), buf[:n]...)}
195+ }
196+ if err != nil {
197+ p.chunks <- readResult{err: err}
198+ close(p.chunks)
199+ return
200+ }
201+ }
202+}
203+
204+func (p *processStdinPump) read(done <-chan struct{}, dst []byte) (int, error) {
205+ select {
206+ case <-done:
207+ return 0, io.ErrClosedPipe
208+ default:
209+ }
210+ p.mu.Lock()
211+ if len(p.buf) != 0 {
212+ n := copy(dst, p.buf)
213+ p.buf = p.buf[n:]
214+ p.mu.Unlock()
215+ return n, nil
216+ }
217+ if p.err != nil {
218+ err := p.err
219+ p.mu.Unlock()
220+ return 0, err
221+ }
222+ p.mu.Unlock()
223+
224+ select {
225+ case <-done:
226+ return 0, io.ErrClosedPipe
227+ case result, ok := <-p.chunks:
228+ if !ok {
229+ return 0, io.EOF
230+ }
231+ if len(result.data) != 0 {
232+ select {
233+ case <-done:
234+ return 0, io.ErrClosedPipe
235+ default:
236+ }
237+ n := copy(dst, result.data)
238+ p.mu.Lock()
239+ p.buf = append(p.buf[:0], result.data[n:]...)
240+ p.mu.Unlock()
241+ return n, nil
242+ }
243+ p.mu.Lock()
244+ p.err = result.err
245+ p.mu.Unlock()
246+ return 0, result.err
247+ }
248+}
249+
250+type processStdinReader struct {
251+ pump *processStdinPump
252+ done <-chan struct{}
253+}
254+
255+func newProcessStdinReader(file *os.File, done <-chan struct{}) io.Reader {
256+ return processStdinReader{pump: processStdinPumpFor(file), done: done}
257+}
258+
259+func (r processStdinReader) Read(dst []byte) (int, error) {
260+ return r.pump.read(r.done, dst)
261+}
262+
92263 func (t *transport) Send(f ports.Frame) error {
93264 end := t.beginOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
94265 n := 1 + len(f.Payload)
@@ -180,24 +351,23 @@ func (t *transport) mapEOFError(err error) error {
180351 func (t *transport) Close() error {
181352 t.closeOnce.Do(func() {
182353 wait := t.beginShutdown()
183- w, writeOwned := t.w.(io.Closer)
184- r, readOwned := t.r.(io.Closer)
185354 // Close owned streams before waiting for in-flight operations. The
186355 // writer is what releases a Send blocked on a child stdin pipe; the
187356 // reader releases a Recv blocked in io.ReadFull. Unowned streams cannot
188- // be interrupted safely, so Close must not wait for their operations.
357+ // be interrupted safely, so the cancellation boundary releases Recv
358+ // without closing the source.
189359 var writeErr, readErr error
190- if writeOwned {
191- writeErr = w.Close()
360+ if t.writerCloser != nil {
361+ writeErr = t.writerCloser.Close()
192362 }
193- if readOwned {
194- readErr = r.Close()
363+ if t.readerCloser != nil {
364+ readErr = t.readerCloser.Close()
195365 }
196366 closeErr := t.close()
197- if writeOwned && wait.send != nil {
367+ if t.writerCloser != nil && wait.send != nil {
198368 <-wait.send
199369 }
200- if readOwned && wait.receive != nil {
370+ if t.readerCloser != nil && wait.receive != nil {
201371 <-wait.receive
202372 }
203373 switch {
@@ -238,6 +408,7 @@ func (t *transport) beginShutdown() operationWait {
238408 t.operationMu.Lock()
239409 defer t.operationMu.Unlock()
240410 t.closing = true
411+ close(t.done)
241412 var wait operationWait
242413 if t.operationCount[operationSend] != 0 {
243414 wait.send = t.operationsDone[operationSend]
@@ -12,6 +12,7 @@ import (
12 "fmt"12 "fmt"
13 "io"13 "io"
14 "log/slog"14 "log/slog"
15+ "os"
15 "os/exec"16 "os/exec"
16 "strings"17 "strings"
17 "sync"18 "sync"
@@ -69,7 +70,26 @@ func newTransport(r io.Reader, w io.Writer, closeFn closeFunc, eofErr eofErrFunc
69 if closeFn == nil {70 if closeFn == nil {
70 closeFn = func() error { return nil }71 closeFn = func() error { return nil }
71 }72 }
72- return &transport{r: r, w: w, close: closeFn, eofErr: eofErr}73+ t := &transport{
74+ r: r,
75+ w: w,
76+ close: closeFn,
77+ eofErr: eofErr,
78+ done: make(chan struct{}),
79+ }
80+ if file, ok := r.(*os.File); ok && file == os.Stdin {
81+ t.r = newProcessStdinReader(file, t.done)
82+ } else if closer, ok := r.(io.Closer); ok {
83+ t.readerCloser = closer
84+ } else {
85+ t.r = newUnownedReader(r, t.done)
86+ }
87+ if file, ok := w.(*os.File); !ok || file != os.Stdout {
88+ if closer, ok := w.(io.Closer); ok {
89+ t.writerCloser = closer
90+ }
91+ }
92+ return t
73 }93 }
74 94
75 type transport struct {95 type transport struct {
@@ -77,6 +97,9 @@ type transport struct {
77 w io.Writer97 w io.Writer
78 close closeFunc98 close closeFunc
79 eofErr eofErrFunc99 eofErr eofErrFunc
100+ readerCloser io.Closer
101+ writerCloser io.Closer
102+ done chan struct{}
80 observer ports.SerializedRuntimeObserver103 observer ports.SerializedRuntimeObserver
81 operationMu sync.Mutex104 operationMu sync.Mutex
82 operationCount [operationKindCount]int105 operationCount [operationKindCount]int
@@ -89,6 +112,154 @@ type transport struct {
89 readBuf []byte112 readBuf []byte
90 }113 }
91 114
115+type readResult struct {
116+ data []byte
117+ err error
118+}
119+
120+// unownedReader is the generic cancellation boundary for readers that cannot
121+// be closed by the transport. The worker may remain blocked until the source
122+// itself is released; process stdin uses the singleton pump below instead.
123+// ponytail: arbitrary io.Reader has no cancellation contract, so one fallback
124+// worker may remain until that source returns; add a cancellable reader port if
125+// another production unowned source appears.
126+type unownedReader struct {
127+ r io.Reader
128+ done <-chan struct{}
129+}
130+
131+func newUnownedReader(r io.Reader, done <-chan struct{}) *unownedReader {
132+ return &unownedReader{r: r, done: done}
133+}
134+
135+func (r *unownedReader) Read(dst []byte) (int, error) {
136+ select {
137+ case <-r.done:
138+ return 0, io.ErrClosedPipe
139+ default:
140+ }
141+ result := make(chan readResult, 1)
142+ go func() {
143+ buf := make([]byte, len(dst))
144+ n, err := r.r.Read(buf)
145+ result <- readResult{data: buf[:n], err: err}
146+ }()
147+ select {
148+ case <-r.done:
149+ return 0, io.ErrClosedPipe
150+ case result := <-result:
151+ select {
152+ case <-r.done:
153+ return 0, io.ErrClosedPipe
154+ default:
155+ }
156+ copy(dst, result.data)
157+ return len(result.data), result.err
158+ }
159+}
160+
161+type processStdinPump struct {
162+ chunks chan readResult
163+ mu sync.Mutex
164+ buf []byte
165+ err error
166+}
167+
168+var processStdinPumps struct {
169+ sync.Mutex
170+ file *os.File
171+ pump *processStdinPump
172+}
173+
174+// processStdinPumpFor keeps one read worker for the process stdin lifetime.
175+// Reconnects therefore do not strand one blocked worker per transport.
176+func processStdinPumpFor(file *os.File) *processStdinPump {
177+ processStdinPumps.Lock()
178+ defer processStdinPumps.Unlock()
179+ if processStdinPumps.pump != nil && processStdinPumps.file == file {
180+ return processStdinPumps.pump
181+ }
182+ pump := &processStdinPump{chunks: make(chan readResult, 1)}
183+ processStdinPumps.file = file
184+ processStdinPumps.pump = pump
185+ go pump.run(file)
186+ return pump
187+}
188+
189+func (p *processStdinPump) run(file *os.File) {
190+ for {
191+ buf := make([]byte, 32*1024)
192+ n, err := file.Read(buf)
193+ if n > 0 {
194+ p.chunks <- readResult{data: append([]byte(nil), buf[:n]...)}
195+ }
196+ if err != nil {
197+ p.chunks <- readResult{err: err}
198+ close(p.chunks)
199+ return
200+ }
201+ }
202+}
203+
204+func (p *processStdinPump) read(done <-chan struct{}, dst []byte) (int, error) {
205+ select {
206+ case <-done:
207+ return 0, io.ErrClosedPipe
208+ default:
209+ }
210+ p.mu.Lock()
211+ if len(p.buf) != 0 {
212+ n := copy(dst, p.buf)
213+ p.buf = p.buf[n:]
214+ p.mu.Unlock()
215+ return n, nil
216+ }
217+ if p.err != nil {
218+ err := p.err
219+ p.mu.Unlock()
220+ return 0, err
221+ }
222+ p.mu.Unlock()
223+
224+ select {
225+ case <-done:
226+ return 0, io.ErrClosedPipe
227+ case result, ok := <-p.chunks:
228+ if !ok {
229+ return 0, io.EOF
230+ }
231+ if len(result.data) != 0 {
232+ select {
233+ case <-done:
234+ return 0, io.ErrClosedPipe
235+ default:
236+ }
237+ n := copy(dst, result.data)
238+ p.mu.Lock()
239+ p.buf = append(p.buf[:0], result.data[n:]...)
240+ p.mu.Unlock()
241+ return n, nil
242+ }
243+ p.mu.Lock()
244+ p.err = result.err
245+ p.mu.Unlock()
246+ return 0, result.err
247+ }
248+}
249+
250+type processStdinReader struct {
251+ pump *processStdinPump
252+ done <-chan struct{}
253+}
254+
255+func newProcessStdinReader(file *os.File, done <-chan struct{}) io.Reader {
256+ return processStdinReader{pump: processStdinPumpFor(file), done: done}
257+}
258+
259+func (r processStdinReader) Read(dst []byte) (int, error) {
260+ return r.pump.read(r.done, dst)
261+}
262+
92 func (t *transport) Send(f ports.Frame) error {263 func (t *transport) Send(f ports.Frame) error {
93 end := t.beginOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))264 end := t.beginOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
94 n := 1 + len(f.Payload)265 n := 1 + len(f.Payload)
@@ -180,24 +351,23 @@ func (t *transport) mapEOFError(err error) error {
180 func (t *transport) Close() error {351 func (t *transport) Close() error {
181 t.closeOnce.Do(func() {352 t.closeOnce.Do(func() {
182 wait := t.beginShutdown()353 wait := t.beginShutdown()
183- w, writeOwned := t.w.(io.Closer)
184- r, readOwned := t.r.(io.Closer)
185 // Close owned streams before waiting for in-flight operations. The354 // Close owned streams before waiting for in-flight operations. The
186 // writer is what releases a Send blocked on a child stdin pipe; the355 // writer is what releases a Send blocked on a child stdin pipe; the
187 // reader releases a Recv blocked in io.ReadFull. Unowned streams cannot356 // reader releases a Recv blocked in io.ReadFull. Unowned streams cannot
188- // be interrupted safely, so Close must not wait for their operations.357+ // be interrupted safely, so the cancellation boundary releases Recv
358+ // without closing the source.
189 var writeErr, readErr error359 var writeErr, readErr error
190- if writeOwned {360+ if t.writerCloser != nil {
191- writeErr = w.Close()361+ writeErr = t.writerCloser.Close()
192 }362 }
193- if readOwned {363+ if t.readerCloser != nil {
194- readErr = r.Close()364+ readErr = t.readerCloser.Close()
195 }365 }
196 closeErr := t.close()366 closeErr := t.close()
197- if writeOwned && wait.send != nil {367+ if t.writerCloser != nil && wait.send != nil {
198 <-wait.send368 <-wait.send
199 }369 }
200- if readOwned && wait.receive != nil {370+ if t.readerCloser != nil && wait.receive != nil {
201 <-wait.receive371 <-wait.receive
202 }372 }
203 switch {373 switch {
@@ -238,6 +408,7 @@ func (t *transport) beginShutdown() operationWait {
238 t.operationMu.Lock()408 t.operationMu.Lock()
239 defer t.operationMu.Unlock()409 defer t.operationMu.Unlock()
240 t.closing = true410 t.closing = true
411+ close(t.done)
241 var wait operationWait412 var wait operationWait
242 if t.operationCount[operationSend] != 0 {413 if t.operationCount[operationSend] != 0 {
243 wait.send = t.operationsDone[operationSend]414 wait.send = t.operationsDone[operationSend]
M internal/adapters/sshstdio/transport_observability_test.go +22 -4
@@ -4,6 +4,7 @@ import (
44 "bytes"
55 "encoding/json"
66 "errors"
7+ "io"
78 "os"
89 "strings"
910 "sync"
@@ -93,7 +94,7 @@ func assertSSHReceiveFailurePair(t *testing.T, path string) {
9394 Kind string `json:"kind"`
9495 Valid bool `json:"valid"`
9596 }
96- for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {
97+ for line := range strings.SplitSeq(strings.TrimSpace(string(data)), "\n") {
9798 var mark struct {
9899 ProcessID string `json:"process_id"`
99100 Scenario string `json:"scenario"`
@@ -126,6 +127,24 @@ func assertSSHReceiveFailurePair(t *testing.T, path string) {
126127 }
127128 }
128129
130+func TestProcessStdinPumpIsSingleton(t *testing.T) {
131+ first := processStdinPumpFor(os.Stdin)
132+ second := processStdinPumpFor(os.Stdin)
133+ if first != second {
134+ t.Fatal("process stdin created more than one lifetime pump")
135+ }
136+}
137+
138+func TestTransportCloseLeavesProcessStdinOpen(t *testing.T) {
139+ transport := NewTransport(os.Stdin, io.Discard, nil)
140+ if err := transport.Close(); err != nil {
141+ t.Fatalf("Close() error = %v", err)
142+ }
143+ if _, err := os.Stdin.Stat(); err != nil {
144+ t.Fatalf("process stdin was closed: %v", err)
145+ }
146+}
147+
129148 func TestTransportObservabilitySSHStdioCloseDoesNotWaitForUnownedReceive(t *testing.T) {
130149 tracePath := t.TempDir() + "/ssh-unowned-close.jsonl"
131150 observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 313)}, "ssh-process")
@@ -152,14 +171,13 @@ func TestTransportObservabilitySSHStdioCloseDoesNotWaitForUnownedReceive(t *test
152171 t.Fatal("Close() waited for an unowned blocked reader")
153172 }
154173
155- reader.unblock()
156174 select {
157175 case err := <-recvDone:
158176 if err == nil {
159- t.Fatal("Recv() error = nil after releasing blocked reader")
177+ t.Fatal("Recv() error = nil after Close()")
160178 }
161179 case <-time.After(time.Second):
162- t.Fatal("Recv() did not finish after releasing blocked reader")
180+ t.Fatal("Recv() did not finish after Close()")
163181 }
164182 reporter.Close()
165183 if err := closer.Close(); err != nil {
@@ -4,6 +4,7 @@ import (
4 "bytes"4 "bytes"
5 "encoding/json"5 "encoding/json"
6 "errors"6 "errors"
7+ "io"
7 "os"8 "os"
8 "strings"9 "strings"
9 "sync"10 "sync"
@@ -93,7 +94,7 @@ func assertSSHReceiveFailurePair(t *testing.T, path string) {
93 Kind string `json:"kind"`94 Kind string `json:"kind"`
94 Valid bool `json:"valid"`95 Valid bool `json:"valid"`
95 }96 }
96- for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {97+ for line := range strings.SplitSeq(strings.TrimSpace(string(data)), "\n") {
97 var mark struct {98 var mark struct {
98 ProcessID string `json:"process_id"`99 ProcessID string `json:"process_id"`
99 Scenario string `json:"scenario"`100 Scenario string `json:"scenario"`
@@ -126,6 +127,24 @@ func assertSSHReceiveFailurePair(t *testing.T, path string) {
126 }127 }
127 }128 }
128 129
130+func TestProcessStdinPumpIsSingleton(t *testing.T) {
131+ first := processStdinPumpFor(os.Stdin)
132+ second := processStdinPumpFor(os.Stdin)
133+ if first != second {
134+ t.Fatal("process stdin created more than one lifetime pump")
135+ }
136+}
137+
138+func TestTransportCloseLeavesProcessStdinOpen(t *testing.T) {
139+ transport := NewTransport(os.Stdin, io.Discard, nil)
140+ if err := transport.Close(); err != nil {
141+ t.Fatalf("Close() error = %v", err)
142+ }
143+ if _, err := os.Stdin.Stat(); err != nil {
144+ t.Fatalf("process stdin was closed: %v", err)
145+ }
146+}
147+
129 func TestTransportObservabilitySSHStdioCloseDoesNotWaitForUnownedReceive(t *testing.T) {148 func TestTransportObservabilitySSHStdioCloseDoesNotWaitForUnownedReceive(t *testing.T) {
130 tracePath := t.TempDir() + "/ssh-unowned-close.jsonl"149 tracePath := t.TempDir() + "/ssh-unowned-close.jsonl"
131 observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 313)}, "ssh-process")150 observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 313)}, "ssh-process")
@@ -152,14 +171,13 @@ func TestTransportObservabilitySSHStdioCloseDoesNotWaitForUnownedReceive(t *test
152 t.Fatal("Close() waited for an unowned blocked reader")171 t.Fatal("Close() waited for an unowned blocked reader")
153 }172 }
154 173
155- reader.unblock()
156 select {174 select {
157 case err := <-recvDone:175 case err := <-recvDone:
158 if err == nil {176 if err == nil {
159- t.Fatal("Recv() error = nil after releasing blocked reader")177+ t.Fatal("Recv() error = nil after Close()")
160 }178 }
161 case <-time.After(time.Second):179 case <-time.After(time.Second):
162- t.Fatal("Recv() did not finish after releasing blocked reader")180+ t.Fatal("Recv() did not finish after Close()")
163 }181 }
164 reporter.Close()182 reporter.Close()
165 if err := closer.Close(); err != nil {183 if err := closer.Close(); err != nil {
M internal/app/run.go +3 -2
@@ -1040,8 +1040,9 @@ func runStdio(ctx context.Context) (retErr error) {
10401040
10411041 // sshstdio.Close owns and closes writers that implement io.Closer so a
10421042 // blocked Send can unwind. Keep process-global stdin/stdout outside that
1043- // ownership boundary: input is a non-closing view and output is an owned
1044- // pipe drained by a small forwarding goroutine.
1043+ // ownership boundary: sshstdio recognizes process stdin as a shared
1044+ // cancellable pump, while output is an owned pipe drained by a forwarding
1045+ // goroutine.
10451046 stdin, stdout := os.Stdin, os.Stdout
10461047 stdoutReader, stdoutWriter := io.Pipe()
10471048 forwardDone := make(chan struct{})
@@ -1040,8 +1040,9 @@ func runStdio(ctx context.Context) (retErr error) {
1040 1040
1041 // sshstdio.Close owns and closes writers that implement io.Closer so a1041 // sshstdio.Close owns and closes writers that implement io.Closer so a
1042 // blocked Send can unwind. Keep process-global stdin/stdout outside that1042 // blocked Send can unwind. Keep process-global stdin/stdout outside that
1043- // ownership boundary: input is a non-closing view and output is an owned1043+ // ownership boundary: sshstdio recognizes process stdin as a shared
1044- // pipe drained by a small forwarding goroutine.1044+ // cancellable pump, while output is an owned pipe drained by a forwarding
1045+ // goroutine.
1045 stdin, stdout := os.Stdin, os.Stdout1046 stdin, stdout := os.Stdin, os.Stdout
1046 stdoutReader, stdoutWriter := io.Pipe()1047 stdoutReader, stdoutWriter := io.Pipe()
1047 forwardDone := make(chan struct{})1048 forwardDone := make(chan struct{})
M internal/usecase/daemon/attachment_render_test.go +65 -0
@@ -31,6 +31,71 @@ func (t *blockedAttachmentTransport) Send(frame ports.Frame) error {
3131 func (*blockedAttachmentTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF }
3232 func (*blockedAttachmentTransport) Close() error { return nil }
3333
34+func TestStaleCapacityReadinessRequeuesWithoutAnotherInvalidation(t *testing.T) {
35+ d, sess, ac, sends := newManualSessionWithPTYs(t, nil)
36+ rc := d.attachCoordinator(sess, nil, ac, true)
37+ rc.opts.clock = nil // drive fire explicitly; the interleaving is the test.
38+
39+ firstObserved := make(chan struct{})
40+ retryObserved := make(chan struct{})
41+ releaseFirst := make(chan struct{})
42+ releaseRetry := make(chan struct{})
43+ var probes atomic.Int32
44+ rc.opts.ackReadyFor = func(*attachedClient) bool {
45+ switch probes.Add(1) {
46+ case 1:
47+ close(firstObserved)
48+ <-releaseFirst
49+ case 2:
50+ close(retryObserved)
51+ <-releaseRetry
52+ }
53+ return true
54+ }
55+
56+ pane := sess.tabs[0].focusedPane()
57+ pane.mu.Lock()
58+ pane.screen.Write([]byte("capacity race"))
59+ pane.mu.Unlock()
60+ require.True(t, rc.invalidate(renderInvalidation{class: invalidateUrgent, producer: "capacity-race-test"}))
61+ fireDone := make(chan struct{})
62+ go func() {
63+ rc.fireCurrent(false)
64+ close(fireDone)
65+ }()
66+ awaitTestCompletion(t, firstObserved, "readiness probe did not start")
67+
68+ // Readiness was observed while the window had room. Fill it before the
69+ // wake reaches paint; paint must report the failed final capacity check.
70+ ac.sendMu.Lock()
71+ ac.output.maxOutstanding = 1
72+ ac.output.next = 1
73+ ac.output.acked = 0
74+ ac.output.syncCapacityLocked()
75+ ac.sendMu.Unlock()
76+ close(releaseFirst)
77+ awaitTestCompletion(t, retryObserved, "capacity failure did not schedule an internal retry")
78+
79+ // The retry is already in flight. Restore capacity without publishing any
80+ // new invalidation or ACK notification, then let that retry paint.
81+ ac.sendMu.Lock()
82+ ac.output.maxOutstanding = 2
83+ ac.output.syncCapacityLocked()
84+ ac.sendMu.Unlock()
85+ close(releaseRetry)
86+ frame := awaitFrame(t, sends, ports.MsgOutput)
87+ output, err := ports.UnmarshalOutput(frame.Payload)
88+ require.NoError(t, err)
89+ require.Contains(t, string(output.Data), "capacity race")
90+ awaitTestCompletion(t, fireDone, "initial fire did not finish")
91+
92+ rc.mu.Lock()
93+ requeuedPending := rc.pending
94+ rc.mu.Unlock()
95+ require.False(t, requeuedPending, "successful internal retry must consume the mutation")
96+ require.Equal(t, int32(2), probes.Load(), "retry must not require another external wake")
97+}
98+
3499 func TestAttachmentResizeKeepsSessionContentAndPeersFixed(t *testing.T) {
35100 pty, releasePTY := newBlockingPTY(t)
36101 defer releasePTY()
@@ -31,6 +31,71 @@ func (t *blockedAttachmentTransport) Send(frame ports.Frame) error {
31 func (*blockedAttachmentTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF }31 func (*blockedAttachmentTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF }
32 func (*blockedAttachmentTransport) Close() error { return nil }32 func (*blockedAttachmentTransport) Close() error { return nil }
33 33
34+func TestStaleCapacityReadinessRequeuesWithoutAnotherInvalidation(t *testing.T) {
35+ d, sess, ac, sends := newManualSessionWithPTYs(t, nil)
36+ rc := d.attachCoordinator(sess, nil, ac, true)
37+ rc.opts.clock = nil // drive fire explicitly; the interleaving is the test.
38+
39+ firstObserved := make(chan struct{})
40+ retryObserved := make(chan struct{})
41+ releaseFirst := make(chan struct{})
42+ releaseRetry := make(chan struct{})
43+ var probes atomic.Int32
44+ rc.opts.ackReadyFor = func(*attachedClient) bool {
45+ switch probes.Add(1) {
46+ case 1:
47+ close(firstObserved)
48+ <-releaseFirst
49+ case 2:
50+ close(retryObserved)
51+ <-releaseRetry
52+ }
53+ return true
54+ }
55+
56+ pane := sess.tabs[0].focusedPane()
57+ pane.mu.Lock()
58+ pane.screen.Write([]byte("capacity race"))
59+ pane.mu.Unlock()
60+ require.True(t, rc.invalidate(renderInvalidation{class: invalidateUrgent, producer: "capacity-race-test"}))
61+ fireDone := make(chan struct{})
62+ go func() {
63+ rc.fireCurrent(false)
64+ close(fireDone)
65+ }()
66+ awaitTestCompletion(t, firstObserved, "readiness probe did not start")
67+
68+ // Readiness was observed while the window had room. Fill it before the
69+ // wake reaches paint; paint must report the failed final capacity check.
70+ ac.sendMu.Lock()
71+ ac.output.maxOutstanding = 1
72+ ac.output.next = 1
73+ ac.output.acked = 0
74+ ac.output.syncCapacityLocked()
75+ ac.sendMu.Unlock()
76+ close(releaseFirst)
77+ awaitTestCompletion(t, retryObserved, "capacity failure did not schedule an internal retry")
78+
79+ // The retry is already in flight. Restore capacity without publishing any
80+ // new invalidation or ACK notification, then let that retry paint.
81+ ac.sendMu.Lock()
82+ ac.output.maxOutstanding = 2
83+ ac.output.syncCapacityLocked()
84+ ac.sendMu.Unlock()
85+ close(releaseRetry)
86+ frame := awaitFrame(t, sends, ports.MsgOutput)
87+ output, err := ports.UnmarshalOutput(frame.Payload)
88+ require.NoError(t, err)
89+ require.Contains(t, string(output.Data), "capacity race")
90+ awaitTestCompletion(t, fireDone, "initial fire did not finish")
91+
92+ rc.mu.Lock()
93+ requeuedPending := rc.pending
94+ rc.mu.Unlock()
95+ require.False(t, requeuedPending, "successful internal retry must consume the mutation")
96+ require.Equal(t, int32(2), probes.Load(), "retry must not require another external wake")
97+}
98+
34 func TestAttachmentResizeKeepsSessionContentAndPeersFixed(t *testing.T) {99 func TestAttachmentResizeKeepsSessionContentAndPeersFixed(t *testing.T) {
35 pty, releasePTY := newBlockingPTY(t)100 pty, releasePTY := newBlockingPTY(t)
36 defer releasePTY()101 defer releasePTY()
M internal/usecase/daemon/client.go +8 -3
@@ -619,12 +619,17 @@ func (d *Daemon) ensureAttachmentRenderCoordinatorPrelocked(entry *session) *ren
619619 // one synchronous exception: its first paint must finish before the
620620 // handshake deadline is released.
621621 fanoutReset := w.reset || len(attachments) > 1
622+ paint := func() {
623+ if d.paint(entry, attachment, fanoutReset, lease) == paintBlockedCapacity {
624+ rc.retryCapacity(w, lease)
625+ }
626+ }
622627 if w.lease != nil && w.lease.attachment == attachment {
623- d.paint(entry, attachment, fanoutReset, lease)
628+ paint()
624629 } else if len(attachments) > 1 {
625- go d.paint(entry, attachment, fanoutReset, lease)
630+ go paint()
626631 } else {
627- d.paint(entry, attachment, fanoutReset, lease)
632+ paint()
628633 }
629634 }
630635 },
@@ -619,12 +619,17 @@ func (d *Daemon) ensureAttachmentRenderCoordinatorPrelocked(entry *session) *ren
619 // one synchronous exception: its first paint must finish before the619 // one synchronous exception: its first paint must finish before the
620 // handshake deadline is released.620 // handshake deadline is released.
621 fanoutReset := w.reset || len(attachments) > 1621 fanoutReset := w.reset || len(attachments) > 1
622+ paint := func() {
623+ if d.paint(entry, attachment, fanoutReset, lease) == paintBlockedCapacity {
624+ rc.retryCapacity(w, lease)
625+ }
626+ }
622 if w.lease != nil && w.lease.attachment == attachment {627 if w.lease != nil && w.lease.attachment == attachment {
623- d.paint(entry, attachment, fanoutReset, lease)628+ paint()
624 } else if len(attachments) > 1 {629 } else if len(attachments) > 1 {
625- go d.paint(entry, attachment, fanoutReset, lease)630+ go paint()
626 } else {631 } else {
627- d.paint(entry, attachment, fanoutReset, lease)632+ paint()
628 }633 }
629 }634 }
630 },635 },
M internal/usecase/daemon/render.go +16 -7
@@ -290,7 +290,15 @@ func (b *runtimeMarkBatch) flush() {
290290 }
291291 }
292292
293-func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *attachmentLease) {
293+type paintResult uint8
294+
295+const (
296+ paintRejected paintResult = iota
297+ paintEmitted
298+ paintBlockedCapacity
299+)
300+
301+func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *attachmentLease) paintResult {
294302 // Session-owned PTY preparation remains local; attachment rendering is
295303 // captured through session.captureRenderState.
296304 sess := entry
@@ -301,7 +309,7 @@ func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *at
301309 token.lease = lease
302310 ticket, admitted := ac.beginAttachmentEffect(token)
303311 if !admitted {
304- return
312+ return paintRejected
305313 }
306314 marks.attachmentEffect = ticket
307315 defer ticket.End()
@@ -310,7 +318,7 @@ func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *at
310318 if local {
311319 tb = sess.tabForAttachment(ac)
312320 if tb == nil {
313- return
321+ return paintRejected
314322 }
315323 }
316324
@@ -323,13 +331,13 @@ func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *at
323331 entry.core().mu.Unlock()
324332 if !owned || ac.currentAttachmentSession() != entry {
325333 ac.sendMu.Unlock()
326- return
334+ return paintRejected
327335 }
328336 if lease != nil {
329337 rc := attachmentRenderCoordinator(entry)
330338 if rc == nil || lease.attachment != ac || !rc.leaseCurrent(lease, true) {
331339 ac.sendMu.Unlock()
332- return
340+ return paintRejected
333341 }
334342 }
335343 // Capacity is checked before any destructive capture. Refresh the atomic
@@ -346,7 +354,7 @@ func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *at
346354 // The coordinator owns the blocked interval; this guard only protects
347355 // direct test/resize paint calls from destructive capture.
348356 ac.sendMu.Unlock()
349- return
357+ return paintBlockedCapacity
350358 }
351359 // Composition owns attachment sendMu; initialize its lazy overlay state
352360 // under that same ownership so concurrent fallback paints cannot observe a
@@ -426,7 +434,7 @@ func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *at
426434 endCapture(0, ok)
427435 if !ok {
428436 ac.sendMu.Unlock()
429- return
437+ return paintRejected
430438 }
431439 if ac.renderStages.capture != nil {
432440 ac.renderStages.capture()
@@ -445,6 +453,7 @@ func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *at
445453 ac.renderStages.compose()
446454 }
447455 d.emitFrame(entry, ac, state, composed, &marks)
456+ return paintEmitted
448457 }
449458
450459 var fallbackChromeStyles = themeui.Resolve(themeui.Theme{}, domain.ThemeAccent{Mode: domain.ThemeAccentAuto}).Styles
@@ -290,7 +290,15 @@ func (b *runtimeMarkBatch) flush() {
290 }290 }
291 }291 }
292 292
293-func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *attachmentLease) {293+type paintResult uint8
294+
295+const (
296+ paintRejected paintResult = iota
297+ paintEmitted
298+ paintBlockedCapacity
299+)
300+
301+func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *attachmentLease) paintResult {
294 // Session-owned PTY preparation remains local; attachment rendering is302 // Session-owned PTY preparation remains local; attachment rendering is
295 // captured through session.captureRenderState.303 // captured through session.captureRenderState.
296 sess := entry304 sess := entry
@@ -301,7 +309,7 @@ func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *at
301 token.lease = lease309 token.lease = lease
302 ticket, admitted := ac.beginAttachmentEffect(token)310 ticket, admitted := ac.beginAttachmentEffect(token)
303 if !admitted {311 if !admitted {
304- return312+ return paintRejected
305 }313 }
306 marks.attachmentEffect = ticket314 marks.attachmentEffect = ticket
307 defer ticket.End()315 defer ticket.End()
@@ -310,7 +318,7 @@ func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *at
310 if local {318 if local {
311 tb = sess.tabForAttachment(ac)319 tb = sess.tabForAttachment(ac)
312 if tb == nil {320 if tb == nil {
313- return321+ return paintRejected
314 }322 }
315 }323 }
316 324
@@ -323,13 +331,13 @@ func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *at
323 entry.core().mu.Unlock()331 entry.core().mu.Unlock()
324 if !owned || ac.currentAttachmentSession() != entry {332 if !owned || ac.currentAttachmentSession() != entry {
325 ac.sendMu.Unlock()333 ac.sendMu.Unlock()
326- return334+ return paintRejected
327 }335 }
328 if lease != nil {336 if lease != nil {
329 rc := attachmentRenderCoordinator(entry)337 rc := attachmentRenderCoordinator(entry)
330 if rc == nil || lease.attachment != ac || !rc.leaseCurrent(lease, true) {338 if rc == nil || lease.attachment != ac || !rc.leaseCurrent(lease, true) {
331 ac.sendMu.Unlock()339 ac.sendMu.Unlock()
332- return340+ return paintRejected
333 }341 }
334 }342 }
335 // Capacity is checked before any destructive capture. Refresh the atomic343 // Capacity is checked before any destructive capture. Refresh the atomic
@@ -346,7 +354,7 @@ func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *at
346 // The coordinator owns the blocked interval; this guard only protects354 // The coordinator owns the blocked interval; this guard only protects
347 // direct test/resize paint calls from destructive capture.355 // direct test/resize paint calls from destructive capture.
348 ac.sendMu.Unlock()356 ac.sendMu.Unlock()
349- return357+ return paintBlockedCapacity
350 }358 }
351 // Composition owns attachment sendMu; initialize its lazy overlay state359 // Composition owns attachment sendMu; initialize its lazy overlay state
352 // under that same ownership so concurrent fallback paints cannot observe a360 // under that same ownership so concurrent fallback paints cannot observe a
@@ -426,7 +434,7 @@ func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *at
426 endCapture(0, ok)434 endCapture(0, ok)
427 if !ok {435 if !ok {
428 ac.sendMu.Unlock()436 ac.sendMu.Unlock()
429- return437+ return paintRejected
430 }438 }
431 if ac.renderStages.capture != nil {439 if ac.renderStages.capture != nil {
432 ac.renderStages.capture()440 ac.renderStages.capture()
@@ -445,6 +453,7 @@ func (d *Daemon) paint(entry *session, ac *attachedClient, reset bool, lease *at
445 ac.renderStages.compose()453 ac.renderStages.compose()
446 }454 }
447 d.emitFrame(entry, ac, state, composed, &marks)455 d.emitFrame(entry, ac, state, composed, &marks)
456+ return paintEmitted
448 }457 }
449 458
450 var fallbackChromeStyles = themeui.Resolve(themeui.Theme{}, domain.ThemeAccent{Mode: domain.ThemeAccentAuto}).Styles459 var fallbackChromeStyles = themeui.Resolve(themeui.Theme{}, domain.ThemeAccent{Mode: domain.ThemeAccentAuto}).Styles
M internal/usecase/daemon/render_coordinator.go +82 -2
@@ -313,6 +313,81 @@ func (c *renderCoordinator) notifyAckForLease(lease *attachmentLease) {
313313 c.fire(gen, false, false)
314314 }
315315
316+// retryCapacity restores only the attachment whose wake failed the final
317+// capacity check. Readiness is an atomic observation, so paint may discover
318+// backpressure after fire has marked the wake delivered. Requeueing the exact
319+// wake generation keeps that stale result from clearing shared work and avoids
320+// making a healthy peer wait for the blocked attachment.
321+func (c *renderCoordinator) retryCapacity(w renderWake, lease *attachmentLease) {
322+ if lease == nil || w.generation == 0 {
323+ return
324+ }
325+ var queueStart bool
326+ c.mu.Lock()
327+ if !c.leaseCurrentLocked(lease, true) || c.pendingGeneration < w.generation {
328+ c.mu.Unlock()
329+ return
330+ }
331+ // A newer wake already attempted this lease; its callback owns any
332+ // capacity retry. An older callback may still widen a newer targeted wake
333+ // to include this lease when that newer wake never reached it.
334+ if c.pendingGeneration > w.generation && lease.deliveredGeneration == c.pendingGeneration {
335+ c.mu.Unlock()
336+ return
337+ }
338+ if !c.pending {
339+ c.pending = true
340+ c.pendingReset = w.reset
341+ c.pendingUrgent = w.urgent
342+ c.pendingShared = false
343+ c.pendingTargets = map[*attachedClient]*attachmentLease{lease.attachment: lease}
344+ c.coalesced = max(1, w.coalesced)
345+ c.deadlineDue = false
346+ c.armed = false
347+ if c.opts.observer != nil && !c.queueMarked {
348+ c.queueMarked = true
349+ c.queueCorrelation = ports.NewRuntimeCorrelation()
350+ queueStart = true
351+ }
352+ } else {
353+ c.pendingReset = c.pendingReset || w.reset
354+ c.pendingUrgent = c.pendingUrgent || w.urgent
355+ if !c.pendingShared {
356+ if c.pendingTargets == nil {
357+ c.pendingTargets = make(map[*attachedClient]*attachmentLease)
358+ }
359+ c.pendingTargets[lease.attachment] = lease
360+ }
361+ }
362+ if c.delivered != nil {
363+ delete(c.delivered, lease.attachment)
364+ }
365+ if c.ackDeferredFor == nil {
366+ c.ackDeferredFor = make(map[*attachedClient]bool)
367+ }
368+ c.ackDeferredFor[lease.attachment] = true
369+ c.ackDeferred = true
370+ generation := c.pendingGeneration
371+ queueCorrelation := c.queueCorrelation
372+ observer := c.opts.observer
373+ c.mu.Unlock()
374+ if queueStart && observer != nil {
375+ observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueEnqueued, 0, true))
376+ }
377+ go c.retryPendingGeneration(generation)
378+}
379+
380+func (c *renderCoordinator) retryPendingGeneration(generation uint64) {
381+ c.mu.Lock()
382+ if c.torndown || !c.pending || c.pendingGeneration != generation {
383+ c.mu.Unlock()
384+ return
385+ }
386+ gen, token := c.normalLane.generation, c.normalLane.token
387+ c.mu.Unlock()
388+ c.fireWithTimerToken(token, gen, false, false)
389+}
390+
316391 // noteSyncBegin records a synchronized-output batch for its stable pane and
317392 // arms that pane's watchdog. Overlapping pane batches are independent.
318393 // markDeadlineDue records a received timer tick before the worker probes ACK
@@ -1010,7 +1085,11 @@ func (c *renderCoordinator) fireWithTimerTokenAndLease(token *timerToken, gen ui
10101085 if len(leases) == 0 && legacyOutput {
10111086 allDelivered = true
10121087 }
1013- w := renderWake{reset: c.pendingReset, urgent: c.pendingUrgent, coalesced: c.coalesced, attachmentLeases: attachmentLeases, watchdog: watchdog}
1088+ wakeGeneration := uint64(0)
1089+ if len(leases) != 0 {
1090+ wakeGeneration = c.pendingGeneration
1091+ }
1092+ w := renderWake{reset: c.pendingReset, urgent: c.pendingUrgent, coalesced: c.coalesced, generation: wakeGeneration, attachmentLeases: attachmentLeases, watchdog: watchdog}
10141093 if syncLease != nil {
10151094 if selected, ok := attachmentLeases[syncLease.attachment]; ok && selected == syncLease {
10161095 w.lease = syncLease
@@ -1040,11 +1119,12 @@ func (c *renderCoordinator) fireWithTimerTokenAndLease(token *timerToken, gen ui
10401119 }
10411120
10421121 if outputReady {
1043- for ac := range attachmentLeases {
1122+ for ac, lease := range attachmentLeases {
10441123 if c.delivered == nil {
10451124 c.delivered = make(map[*attachedClient]uint64)
10461125 }
10471126 c.delivered[ac] = c.pendingGeneration
1127+ lease.deliveredGeneration = c.pendingGeneration
10481128 }
10491129 }
10501130 if len(leases) != 0 {
@@ -313,6 +313,81 @@ func (c *renderCoordinator) notifyAckForLease(lease *attachmentLease) {
313 c.fire(gen, false, false)313 c.fire(gen, false, false)
314 }314 }
315 315
316+// retryCapacity restores only the attachment whose wake failed the final
317+// capacity check. Readiness is an atomic observation, so paint may discover
318+// backpressure after fire has marked the wake delivered. Requeueing the exact
319+// wake generation keeps that stale result from clearing shared work and avoids
320+// making a healthy peer wait for the blocked attachment.
321+func (c *renderCoordinator) retryCapacity(w renderWake, lease *attachmentLease) {
322+ if lease == nil || w.generation == 0 {
323+ return
324+ }
325+ var queueStart bool
326+ c.mu.Lock()
327+ if !c.leaseCurrentLocked(lease, true) || c.pendingGeneration < w.generation {
328+ c.mu.Unlock()
329+ return
330+ }
331+ // A newer wake already attempted this lease; its callback owns any
332+ // capacity retry. An older callback may still widen a newer targeted wake
333+ // to include this lease when that newer wake never reached it.
334+ if c.pendingGeneration > w.generation && lease.deliveredGeneration == c.pendingGeneration {
335+ c.mu.Unlock()
336+ return
337+ }
338+ if !c.pending {
339+ c.pending = true
340+ c.pendingReset = w.reset
341+ c.pendingUrgent = w.urgent
342+ c.pendingShared = false
343+ c.pendingTargets = map[*attachedClient]*attachmentLease{lease.attachment: lease}
344+ c.coalesced = max(1, w.coalesced)
345+ c.deadlineDue = false
346+ c.armed = false
347+ if c.opts.observer != nil && !c.queueMarked {
348+ c.queueMarked = true
349+ c.queueCorrelation = ports.NewRuntimeCorrelation()
350+ queueStart = true
351+ }
352+ } else {
353+ c.pendingReset = c.pendingReset || w.reset
354+ c.pendingUrgent = c.pendingUrgent || w.urgent
355+ if !c.pendingShared {
356+ if c.pendingTargets == nil {
357+ c.pendingTargets = make(map[*attachedClient]*attachmentLease)
358+ }
359+ c.pendingTargets[lease.attachment] = lease
360+ }
361+ }
362+ if c.delivered != nil {
363+ delete(c.delivered, lease.attachment)
364+ }
365+ if c.ackDeferredFor == nil {
366+ c.ackDeferredFor = make(map[*attachedClient]bool)
367+ }
368+ c.ackDeferredFor[lease.attachment] = true
369+ c.ackDeferred = true
370+ generation := c.pendingGeneration
371+ queueCorrelation := c.queueCorrelation
372+ observer := c.opts.observer
373+ c.mu.Unlock()
374+ if queueStart && observer != nil {
375+ observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueEnqueued, 0, true))
376+ }
377+ go c.retryPendingGeneration(generation)
378+}
379+
380+func (c *renderCoordinator) retryPendingGeneration(generation uint64) {
381+ c.mu.Lock()
382+ if c.torndown || !c.pending || c.pendingGeneration != generation {
383+ c.mu.Unlock()
384+ return
385+ }
386+ gen, token := c.normalLane.generation, c.normalLane.token
387+ c.mu.Unlock()
388+ c.fireWithTimerToken(token, gen, false, false)
389+}
390+
316 // noteSyncBegin records a synchronized-output batch for its stable pane and391 // noteSyncBegin records a synchronized-output batch for its stable pane and
317 // arms that pane's watchdog. Overlapping pane batches are independent.392 // arms that pane's watchdog. Overlapping pane batches are independent.
318 // markDeadlineDue records a received timer tick before the worker probes ACK393 // markDeadlineDue records a received timer tick before the worker probes ACK
@@ -1010,7 +1085,11 @@ func (c *renderCoordinator) fireWithTimerTokenAndLease(token *timerToken, gen ui
1010 if len(leases) == 0 && legacyOutput {1085 if len(leases) == 0 && legacyOutput {
1011 allDelivered = true1086 allDelivered = true
1012 }1087 }
1013- w := renderWake{reset: c.pendingReset, urgent: c.pendingUrgent, coalesced: c.coalesced, attachmentLeases: attachmentLeases, watchdog: watchdog}1088+ wakeGeneration := uint64(0)
1089+ if len(leases) != 0 {
1090+ wakeGeneration = c.pendingGeneration
1091+ }
1092+ w := renderWake{reset: c.pendingReset, urgent: c.pendingUrgent, coalesced: c.coalesced, generation: wakeGeneration, attachmentLeases: attachmentLeases, watchdog: watchdog}
1014 if syncLease != nil {1093 if syncLease != nil {
1015 if selected, ok := attachmentLeases[syncLease.attachment]; ok && selected == syncLease {1094 if selected, ok := attachmentLeases[syncLease.attachment]; ok && selected == syncLease {
1016 w.lease = syncLease1095 w.lease = syncLease
@@ -1040,11 +1119,12 @@ func (c *renderCoordinator) fireWithTimerTokenAndLease(token *timerToken, gen ui
1040 }1119 }
1041 1120
1042 if outputReady {1121 if outputReady {
1043- for ac := range attachmentLeases {1122+ for ac, lease := range attachmentLeases {
1044 if c.delivered == nil {1123 if c.delivered == nil {
1045 c.delivered = make(map[*attachedClient]uint64)1124 c.delivered = make(map[*attachedClient]uint64)
1046 }1125 }
1047 c.delivered[ac] = c.pendingGeneration1126 c.delivered[ac] = c.pendingGeneration
1127+ lease.deliveredGeneration = c.pendingGeneration
1048 }1128 }
1049 }1129 }
1050 if len(leases) != 0 {1130 if len(leases) != 0 {
M internal/usecase/daemon/render_coordinator_test.go +1 -0
@@ -195,6 +195,7 @@ func awaitWake(t *testing.T, ch chan renderWake) renderWake {
195195 t.Helper()
196196 wake := awaitTestValue(t, ch, "coordinator did not publish a wake after fake-clock advancement")
197197 wake.attachmentLeases = nil
198+ wake.generation = 0
198199 return wake
199200 }
200201
@@ -195,6 +195,7 @@ func awaitWake(t *testing.T, ch chan renderWake) renderWake {
195 t.Helper()195 t.Helper()
196 wake := awaitTestValue(t, ch, "coordinator did not publish a wake after fake-clock advancement")196 wake := awaitTestValue(t, ch, "coordinator did not publish a wake after fake-clock advancement")
197 wake.attachmentLeases = nil197 wake.attachmentLeases = nil
198+ wake.generation = 0
198 return wake199 return wake
199 }200 }
200 201
M internal/usecase/daemon/render_coordinator_types.go +9 -2
@@ -51,6 +51,9 @@ type renderWake struct {
5151 reset bool
5252 urgent bool
5353 coalesced int
54+ // generation fences a capacity retry to the mutation that produced this
55+ // wake. A newer invalidation owns its own retry path.
56+ generation uint64
5457 // lease is optional coordinator-owned lifecycle safety for an attachment
5558 // specific wake. Shared wakes leave it nil and snapshot live attachment
5659 // leases at dispatch time; each paint then revalidates its exact connection
@@ -68,8 +71,12 @@ type attachmentLease struct {
6871 attachment *attachedClient
6972 ready bool
7073 active bool
71- // A lease is an immutable callback capability. Its lifecycle bits are
72- // changed only under renderCoordinator.mu; callbacks revalidate it at each
74+ // deliveredGeneration records the latest wake that reached this lease. It
75+ // outlives the coordinator's coalesced delivered map so an older capacity
76+ // failure cannot overwrite a newer attempt.
77+ deliveredGeneration uint64
78+ // A lease is an immutable callback capability. Its lifecycle and delivery
79+ // bits are changed only under renderCoordinator.mu; callbacks revalidate it at each
7380 // short effect boundary and never retain coordinator ownership while routing
7481 // arbitrary handlers.
7582 }
@@ -51,6 +51,9 @@ type renderWake struct {
51 reset bool51 reset bool
52 urgent bool52 urgent bool
53 coalesced int53 coalesced int
54+ // generation fences a capacity retry to the mutation that produced this
55+ // wake. A newer invalidation owns its own retry path.
56+ generation uint64
54 // lease is optional coordinator-owned lifecycle safety for an attachment57 // lease is optional coordinator-owned lifecycle safety for an attachment
55 // specific wake. Shared wakes leave it nil and snapshot live attachment58 // specific wake. Shared wakes leave it nil and snapshot live attachment
56 // leases at dispatch time; each paint then revalidates its exact connection59 // leases at dispatch time; each paint then revalidates its exact connection
@@ -68,8 +71,12 @@ type attachmentLease struct {
68 attachment *attachedClient71 attachment *attachedClient
69 ready bool72 ready bool
70 active bool73 active bool
71- // A lease is an immutable callback capability. Its lifecycle bits are74+ // deliveredGeneration records the latest wake that reached this lease. It
72- // changed only under renderCoordinator.mu; callbacks revalidate it at each75+ // outlives the coordinator's coalesced delivered map so an older capacity
76+ // failure cannot overwrite a newer attempt.
77+ deliveredGeneration uint64
78+ // A lease is an immutable callback capability. Its lifecycle and delivery
79+ // bits are changed only under renderCoordinator.mu; callbacks revalidate it at each
73 // short effect boundary and never retain coordinator ownership while routing80 // short effect boundary and never retain coordinator ownership while routing
74 // arbitrary handlers.81 // arbitrary handlers.
75 }82 }

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 2

Tags 13

↑↓ moveenter openesc close