detached 0b11e7b

fix(proxy): align byte-stream handshake

commit
0b11e7b5f65b546a6266e6a332a98a5c7b4ff2c7
author
brice · · 2026-08-04 03:48 UTC
parent
560c7f3
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
3 files · +6 -595
Unified
UnifiedSplit
M internal/usecase/daemon/proxy_link.go +2 -3
@@ -96,6 +96,7 @@ func (d *Daemon) dialProxyHandshake(parent context.Context, p *proxySession, int
9696 p.mu.Lock()
9797 resumeToken := p.resumeToken
9898 clientID := p.clientID
99+ size := p.contentSize
99100 p.mu.Unlock()
100101 resume := intent == ports.IntentResume
101102 if !resume {
@@ -120,6 +121,7 @@ func (d *Daemon) dialProxyHandshake(parent context.Context, p *proxySession, int
120121 ClientID: clientID,
121122 ResumeToken: resumeToken,
122123 Name: p.key.Name,
124+ Size: size,
123125 MaxOutputInFlight: proxyOutputWindow(transport),
124126 }
125127 if err := p.sendGeneration(generation, ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}); err != nil {
@@ -240,9 +242,6 @@ func validateProxyWelcome(frame ports.Frame, sessionName string) (ports.Welcome,
240242 if err != nil {
241243 return ports.Welcome{}, fmt.Errorf("proxy session: decode welcome: %w", err)
242244 }
243- if welcome.Capabilities&ports.CapabilityProxied == 0 {
244- return ports.Welcome{}, errors.New("proxy session: remote did not negotiate proxied capability")
245- }
246245 if welcome.SessionID == "" || welcome.SessionName != sessionName {
247246 return ports.Welcome{}, fmt.Errorf("proxy session: welcome identity mismatch: got %q", welcome.SessionName)
248247 }
@@ -96,6 +96,7 @@ func (d *Daemon) dialProxyHandshake(parent context.Context, p *proxySession, int
96 p.mu.Lock()96 p.mu.Lock()
97 resumeToken := p.resumeToken97 resumeToken := p.resumeToken
98 clientID := p.clientID98 clientID := p.clientID
99+ size := p.contentSize
99 p.mu.Unlock()100 p.mu.Unlock()
100 resume := intent == ports.IntentResume101 resume := intent == ports.IntentResume
101 if !resume {102 if !resume {
@@ -120,6 +121,7 @@ func (d *Daemon) dialProxyHandshake(parent context.Context, p *proxySession, int
120 ClientID: clientID,121 ClientID: clientID,
121 ResumeToken: resumeToken,122 ResumeToken: resumeToken,
122 Name: p.key.Name,123 Name: p.key.Name,
124+ Size: size,
123 MaxOutputInFlight: proxyOutputWindow(transport),125 MaxOutputInFlight: proxyOutputWindow(transport),
124 }126 }
125 if err := p.sendGeneration(generation, ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}); err != nil {127 if err := p.sendGeneration(generation, ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}); err != nil {
@@ -240,9 +242,6 @@ func validateProxyWelcome(frame ports.Frame, sessionName string) (ports.Welcome,
240 if err != nil {242 if err != nil {
241 return ports.Welcome{}, fmt.Errorf("proxy session: decode welcome: %w", err)243 return ports.Welcome{}, fmt.Errorf("proxy session: decode welcome: %w", err)
242 }244 }
243- if welcome.Capabilities&ports.CapabilityProxied == 0 {
244- return ports.Welcome{}, errors.New("proxy session: remote did not negotiate proxied capability")
245- }
246 if welcome.SessionID == "" || welcome.SessionName != sessionName {245 if welcome.SessionID == "" || welcome.SessionName != sessionName {
247 return ports.Welcome{}, fmt.Errorf("proxy session: welcome identity mismatch: got %q", welcome.SessionName)246 return ports.Welcome{}, fmt.Errorf("proxy session: welcome identity mismatch: got %q", welcome.SessionName)
248 }247 }
M internal/usecase/daemon/proxy_output_test.go +3 -590
@@ -1,600 +1,13 @@
11 package daemon
22
33 import (
4- "context"
5- "errors"
6- "fmt"
7- "io"
8- "time"
4+ "testing"
95
6+ "github.com/bnema/vev/internal/domain"
107 "github.com/bnema/vev/internal/ports"
8+ "github.com/stretchr/testify/require"
119 )
1210
13-const (
14- proxyAttachedCommandTimeout = 10 * time.Second
15- proxyResumeInitialBackoff = 250 * time.Millisecond
16- proxyResumeMaxBackoff = 4 * time.Second
17- proxyResumeMaxAttempts = 5
18- proxyResumeStableDuration = 30 * time.Second
19-)
20-
21-var (
22- errProxyCommandTimeout = errors.New("proxy session: attached command timed out")
23- errProxyCommandUnavailable = errors.New("proxy session: attached command link is unavailable")
24- errProxyCommandGeneration = errors.New("proxy session: attached command link was replaced")
25- // errProxyLinkSend marks a reply that could not be written to the remote.
26- // The received frame was well formed, so the link is resumable rather than
27- // terminally broken.
28- errProxyLinkSend = errors.New("proxy session: link reply send failed")
29-)
30-
31-type proxyCommandOutcome struct {
32- result ports.CommandResult
33- err error
34-}
35-
36-type proxyCommandPending struct {
37- generation uint64
38- outcome chan proxyCommandOutcome
39-}
40-
41-type proxyRecvResult struct {
42- frame ports.Frame
43- err error
44-}
45-
46-type proxyLinkResult uint8
47-
48-const (
49- proxyLinkStop proxyLinkResult = iota
50- proxyLinkResume
51- proxyLinkReplaced
52-)
53-
54-// dialProxyHandshake installs a candidate transport, sends an ordinary Hello
55-// through the session's sole sender, and requires a matching Welcome. The
56-// candidate is never registry-visible.
57-func (d *Daemon) dialProxyHandshake(ctx context.Context, p *proxySession, intent uint8) error {
58- if err := ctx.Err(); err != nil {
59- return err
60- }
61- dialer, err := d.remoteDialerFactory.DialerForRemote(p.key.Host, p.key.Name, d.remoteTransportMode, d.log)
62- if err != nil {
63- return fmt.Errorf("proxy session: select remote dialer: %w", err)
64- }
65- if dialer == nil {
66- return errors.New("proxy session: remote dialer factory returned nil dialer")
67- }
68- if err := ctx.Err(); err != nil {
69- return err
70- }
71- transport, err := dialer.Dial(ctx)
72- if err != nil {
73- return fmt.Errorf("proxy session: dial: %w", err)
74- }
75- if transport == nil {
76- return errors.New("proxy session: remote dialer returned nil transport")
77- }
78- if err := ctx.Err(); err != nil {
79- _ = transport.Close()
80- return err
81- }
82-
83- p.mu.Lock()
84- resumeToken := p.resumeToken
85- clientID := p.clientID
86- size := p.contentSize
87- p.mu.Unlock()
88- resume := intent == ports.IntentResume
89- if !resume {
90- resumeToken = 0
91- }
92- generation, previous := p.installTransport(transport, resume)
93- if previous != nil && previous != transport {
94- _ = previous.Close()
95- }
96- failed := true
97- defer func() {
98- if failed {
99- p.retireTransport(generation, transport)
100- _ = transport.Close()
101- }
102- }()
103- hello := ports.Hello{
104- Version: ports.ProtocolVersion,
105- Intent: intent,
106- ClientID: clientID,
107- ResumeToken: resumeToken,
108- Name: p.key.Name,
109- Size: size,
110- MaxOutputInFlight: proxyOutputWindow(transport),
111- }
112- if err := p.sendGeneration(generation, ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}); err != nil {
113- return fmt.Errorf("proxy session: send hello: %w", err)
114- }
115-
116- welcome, err := recvProxyHandshake(ctx, transport, p.key.Name)
117- if err != nil {
118- return err
119- }
120- if err := ctx.Err(); err != nil {
121- return err
122- }
123-
124- p.mu.Lock()
125- if p.linkGeneration != generation || p.transport != transport {
126- p.mu.Unlock()
127- return errors.New("proxy session: link replaced during handshake")
128- }
129- p.resumeToken = welcome.ResumeToken
130- p.linkState = ports.LinkStateConnected
131- p.expired = false
132- p.mu.Unlock()
133- failed = false
134- return nil
135-}
136-
137-func proxyOutputWindow(transport ports.Transport) uint8 {
138- if _, ok := transport.(ports.DatagramTransport); ok {
139- return 1
140- }
141- return maxUnackedOutputStates
142-}
143-
144-func recvProxyHandshake(ctx context.Context, transport ports.Transport, sessionName string) (ports.Welcome, error) {
145- frame, err := recvProxyHandshakeFrame(ctx, transport)
146- if err != nil {
147- return ports.Welcome{}, fmt.Errorf("proxy session: receive welcome: %w", err)
148- }
149- welcome, err := validateProxyWelcome(frame, sessionName)
150- if err != nil {
151- return ports.Welcome{}, err
152- }
153- return welcome, nil
154-}
155-
156-func validateProxyWelcome(frame ports.Frame, sessionName string) (ports.Welcome, error) {
157- if frame.Type == ports.MsgError {
158- remoteErr, err := ports.UnmarshalErrorMsg(frame.Payload)
159- if err != nil {
160- return ports.Welcome{}, fmt.Errorf("proxy session: malformed remote error: %w", err)
161- }
162- return ports.Welcome{}, fmt.Errorf("proxy session: remote rejected handshake (%d): %s", remoteErr.Code, remoteErr.Text)
163- }
164- if frame.Type != ports.MsgWelcome {
165- return ports.Welcome{}, fmt.Errorf("proxy session: first server frame is %d, want Welcome", frame.Type)
166- }
167- welcome, err := ports.UnmarshalWelcome(frame.Payload)
168- if err != nil {
169- return ports.Welcome{}, fmt.Errorf("proxy session: decode welcome: %w", err)
170- }
171- if welcome.SessionID == "" || welcome.SessionName != sessionName {
172- return ports.Welcome{}, fmt.Errorf("proxy session: welcome identity mismatch: got %q", welcome.SessionName)
173- }
174- return welcome, nil
175-}
176-
177-func recvProxyHandshakeFrame(ctx context.Context, transport ports.Transport) (ports.Frame, error) {
178- result := make(chan proxyRecvResult, 1)
179- go func() {
180- frame, err := transport.Recv()
181- result <- proxyRecvResult{frame: frame, err: err}
182- }()
183- select {
184- case received := <-result:
185- return received.frame, received.err
186- case <-ctx.Done():
187- _ = transport.Close()
188- received := <-result // Transport.Close is required to unblock Recv.
189- if received.err != nil && !errors.Is(received.err, io.EOF) {
190- return ports.Frame{}, errors.Join(ctx.Err(), received.err)
191- }
192- return ports.Frame{}, ctx.Err()
193- }
194-}
195-
196-func (p *proxySession) installTransport(transport ports.Transport, _ bool) (uint64, ports.Transport) {
197- // Link replacement participates in the same sender serialization as every
198- // frame. A request can therefore only be published wholly before the
199- // generation changes or wholly after it.
200- p.sendMu.Lock()
201- defer p.sendMu.Unlock()
202- p.mu.Lock()
203- previous := p.transport
204- p.linkGeneration++
205- generation := p.linkGeneration
206- // Each transport must establish its ordinary output dependency with a
207- // dependency-free full frame before incremental bytes are forwarded.
208- p.resetOutputStateLocked()
209- p.transport = transport
210- // Authentication is complete after the remote Welcome; ordinary output
211- // establishes the dependency chain asynchronously.
212- p.linkState = ports.LinkStateProbing
213- p.mu.Unlock()
214- p.failPendingCommandsLocked(errProxyCommandGeneration)
215- return generation, previous
216-}
217-
218-func (p *proxySession) retireTransport(generation uint64, transport ports.Transport) {
219- p.sendMu.Lock()
220- defer p.sendMu.Unlock()
221- p.mu.Lock()
222- current := p.linkGeneration == generation && p.transport == transport
223- if current {
224- p.transport = nil
225- }
226- p.mu.Unlock()
227- if current {
228- p.failPendingCommandsLocked(errProxyCommandUnavailable)
229- }
230-}
231-
232-func (p *proxySession) sendGeneration(generation uint64, frame ports.Frame) error {
233- p.sendMu.Lock()
234- defer p.sendMu.Unlock()
235- p.mu.Lock()
236- if p.linkGeneration != generation || p.transport == nil {
237- p.mu.Unlock()
238- return errors.New("proxy session: stale or unavailable link")
239- }
240- transport := p.transport
241- p.mu.Unlock()
242- return transport.Send(frame)
243-}
244-
245-// sendCommand sends one correlated attached palette command and waits for its
246-// exact result. commandMu serializes interactive requests while sendMu remains
247-// available to all other link traffic during the wait.
248-func (p *proxySession) sendCommand(ctx context.Context, clock ports.Clock, slug string, args []string) (ports.CommandResult, error) {
249- if p == nil || clock == nil {
250- return ports.CommandResult{}, errProxyCommandUnavailable
251- }
252- if ctx == nil {
253- ctx = context.Background()
254- }
255- if err := ctx.Err(); err != nil {
256- return ports.CommandResult{}, err
257- }
258- p.commandMu.Lock()
259- defer p.commandMu.Unlock()
260- if err := ctx.Err(); err != nil {
261- return ports.CommandResult{}, err
262- }
263-
264- p.sendMu.Lock()
265- p.mu.Lock()
266- generation, transport := p.linkGeneration, p.transport
267- p.mu.Unlock()
268- if transport == nil {
269- p.sendMu.Unlock()
270- return ports.CommandResult{}, errProxyCommandUnavailable
271- }
272- p.commandNext++
273- if p.commandNext == 0 {
274- p.commandNext++
275- }
276- requestID := p.commandNext
277- payload, err := ports.MarshalCommandRequest(ports.CommandRequest{
278- Version: ports.ProtocolVersion, RequestID: requestID, Attached: true,
279- Slug: slug, Args: append([]string(nil), args...),
280- })
281- if err != nil {
282- p.sendMu.Unlock()
283- return ports.CommandResult{}, err
284- }
285- outcome := make(chan proxyCommandOutcome, 1)
286- p.commandPending[requestID] = proxyCommandPending{generation: generation, outcome: outcome}
287- if err := transport.Send(ports.Frame{Type: ports.MsgCommand, Payload: payload}); err != nil {
288- delete(p.commandPending, requestID)
289- p.sendMu.Unlock()
290- return ports.CommandResult{}, err
291- }
292- p.sendMu.Unlock()
293-
294- timer := clock.NewTimer(proxyAttachedCommandTimeout)
295- if timer == nil {
296- p.removePendingCommand(requestID, generation)
297- return ports.CommandResult{}, errProxyCommandTimeout
298- }
299- defer timer.Stop()
300- select {
301- case completed := <-outcome:
302- return completed.result, completed.err
303- case <-ctx.Done():
304- p.removePendingCommand(requestID, generation)
305- return ports.CommandResult{}, ctx.Err()
306- case <-timer.C():
307- p.removePendingCommand(requestID, generation)
308- return ports.CommandResult{}, errProxyCommandTimeout
309- }
310-}
311-
312-func (p *proxySession) removePendingCommand(requestID, generation uint64) {
313- p.sendMu.Lock()
314- pending, ok := p.commandPending[requestID]
315- if ok && pending.generation == generation {
316- delete(p.commandPending, requestID)
317- }
318- p.sendMu.Unlock()
319-}
320-
321-// completeCommandResult ignores malformed correlation state, unknown/late IDs,
322-// and results received by a replacement generation.
323-func (p *proxySession) completeCommandResult(generation uint64, result ports.CommandResult) {
324- if result.RequestID == 0 {
325- return
326- }
327- p.sendMu.Lock()
328- pending, ok := p.commandPending[result.RequestID]
329- if !ok || pending.generation != generation {
330- p.sendMu.Unlock()
331- return
332- }
333- delete(p.commandPending, result.RequestID)
334- p.sendMu.Unlock()
335- pending.outcome <- proxyCommandOutcome{result: result}
336-}
337-
338-// failPendingCommandsLocked requires sendMu and never blocks.
339-func (p *proxySession) failPendingCommandsLocked(err error) {
340- for id, pending := range p.commandPending {
341- delete(p.commandPending, id)
342- pending.outcome <- proxyCommandOutcome{err: err}
343- }
344-}
345-
346-func (d *Daemon) runProxyLink(ctx context.Context, p *proxySession) {
347- defer p.finish()
348- connectedAt := d.clock.Now()
349- resumeAttempts := 0
350- for {
351- p.mu.Lock()
352- generation := p.linkGeneration
353- transport := p.transport
354- p.mu.Unlock()
355- if transport == nil {
356- return
357- }
358-
359- result := d.runProxyTransport(ctx, p, generation, transport)
360- if result == proxyLinkReplaced {
361- // ReasonReplaced is terminal for the exact transport incarnation. Publish
362- // expiry before retirement clears the transport identity used by the
363- // callback fence; the switch below deliberately never redials it.
364- d.markProxyReplaced(p, generation, transport)
365- }
366- p.retireTransport(generation, transport)
367- switch result {
368- case proxyLinkReplaced, proxyLinkStop:
369- return
370- case proxyLinkResume:
371- if ctx.Err() != nil {
372- return
373- }
374- if d.clock.Now().Sub(connectedAt) >= proxyResumeStableDuration {
375- resumeAttempts = 0
376- }
377- d.updateProxyDisconnectedState(p, generation, ports.LinkStateOffline)
378-
379- for {
380- if resumeAttempts >= proxyResumeMaxAttempts {
381- d.log.Warn("proxy resume attempts exhausted", "host", p.key.Host, "session", p.key.Name, "attempts", resumeAttempts)
382- return
383- }
384- delay := proxyResumeBackoff(resumeAttempts)
385- if !waitProxyResume(ctx, d.clock, delay) {
386- return
387- }
388- resumeAttempts++
389- if err := d.dialProxyHandshake(ctx, p, ports.IntentResume); err != nil {
390- p.mu.Lock()
391- failedGeneration := p.linkGeneration
392- p.mu.Unlock()
393- d.updateProxyDisconnectedState(p, failedGeneration, ports.LinkStateOffline)
394- if ctx.Err() != nil {
395- return
396- }
397- d.log.Warn("proxy resume failed", "host", p.key.Host, "session", p.key.Name, "attempt", resumeAttempts, "err", err)
398- continue
399- }
400- connectedAt = d.clock.Now()
401- d.repaintProxyLifecycle(p)
402- break
403- }
404- }
405- }
406-}
407-
408-func proxyResumeBackoff(attempt int) time.Duration {
409- delay := proxyResumeInitialBackoff
410- for range attempt {
411- if delay >= proxyResumeMaxBackoff/2 {
412- return proxyResumeMaxBackoff
413- }
414- delay *= 2
415- }
416- return min(delay, proxyResumeMaxBackoff)
417-}
418-
419-func waitProxyResume(ctx context.Context, clock ports.Clock, delay time.Duration) bool {
420- if err := ctx.Err(); err != nil {
421- return false
422- }
423- timer := clock.NewTimer(delay)
424- if timer == nil {
425- return false
426- }
427- defer timer.Stop()
428- select {
429- case <-ctx.Done():
430- return false
431- case <-timer.C():
432- return true
433- }
434-}
435-
436-func (d *Daemon) runProxyTransport(ctx context.Context, p *proxySession, generation uint64, transport ports.Transport) proxyLinkResult {
437- recvCtx, cancelRecv := context.WithCancel(ctx)
438- recv := make(chan proxyRecvResult, 1)
439- recvDone := make(chan struct{})
440- go func() {
441- defer close(recvDone)
442- for {
443- frame, err := transport.Recv()
444- select {
445- case recv <- proxyRecvResult{frame: frame, err: err}:
446- case <-recvCtx.Done():
447- return
448- }
449- if err != nil {
450- return
451- }
452- }
453- }()
454- defer func() {
455- cancelRecv()
456- _ = transport.Close()
457- <-recvDone
458- }()
459-
460- var events <-chan ports.LinkEvent
461- if reporter, ok := transport.(ports.LinkStateReporter); ok {
462- events = reporter.LinkEvents()
463- }
464- for {
465- select {
466- case <-ctx.Done():
467- return proxyLinkStop
468- case event, ok := <-events:
469- if !ok {
470- events = nil
471- continue
472- }
473- if !d.updateProxyLinkState(p, generation, transport, event.State) {
474- return proxyLinkStop
475- }
476- if event.State == ports.LinkStateOffline || event.State == ports.LinkStateDead {
477- _ = transport.Close()
478- return proxyLinkResume
479- }
480- case received := <-recv:
481- if received.err != nil {
482- if ctx.Err() != nil {
483- return proxyLinkStop
484- }
485- return proxyLinkResume
486- }
487- result, err := d.handleLinkFrame(p, generation, received.frame)
488- if err != nil {
489- // A failed reply says nothing about the remote's framing, so it
490- // retires this transport instead of the whole proxy link.
491- if errors.Is(err, errProxyLinkSend) {
492- d.log.Warn("proxy link reply failed", "host", p.key.Host, "session", p.key.Name, "err", err)
493- return proxyLinkResume
494- }
495- d.log.Warn("invalid proxy link frame", "host", p.key.Host, "session", p.key.Name, "err", err)
496- return proxyLinkStop
497- }
498- if result != proxyLinkResume {
499- return result
500- }
501- }
502- }
503-}
504-
505-// handleProxySideEffect forwards a stateless remote effect to the exact local
506-// attachment that currently owns this proxy. Connection admission and transport
507-// incarnation checks make a stale handoff a harmless drop rather than a send
508-// to the client's next session.
509-func (d *Daemon) handleProxySideEffect(p *proxySession, generation uint64, out ports.Output) error {
510- if p == nil || !p.currentLinkGeneration(generation) {
511- return nil
512- }
513- attachments := snapshotAttachmentSession(p)
514- if len(attachments) == 0 {
515- return nil
516- }
517- ac := attachments[0]
518- if ac.output == nil {
519- return nil
520- }
521- expected := ac.transportSnapshot()
522- if expected.transport == nil {
523- return nil
524- }
525- token := attachmentToken(p, ac, expected.transport)
526- if token.sess != p {
527- return nil
528- }
529- ticket, admitted := ac.beginAttachmentEffect(token)
530- if !admitted {
531- return nil
532- }
533- defer ticket.End()
534- if !p.currentLinkGeneration(generation) {
535- return nil
536- }
537- frame := ac.output.sideEffect(out.Data, ac.echoAck.Load())
538- if err := ac.sendExpectedTransportForAttachment(expected, frame, ticket); err != nil {
539- if errors.Is(err, errAttachmentTransition) {
540- return nil
541- }
542- return fmt.Errorf("proxy session: side-effect send: %w: %w", errProxyLinkSend, err)
543- }
544- return nil
545-}
546-
547-// handleLinkFrame returns proxyLinkResume to mean "continue this transport";
548-// the other values are terminal actions.
549-func (d *Daemon) handleLinkFrame(p *proxySession, generation uint64, frame ports.Frame) (proxyLinkResult, error) {
550- switch frame.Type {
551- case ports.MsgOutput:
552- out, err := ports.UnmarshalOutput(frame.Payload)
553- if err != nil {
554- return proxyLinkStop, err
555- }
556- if err := d.handleProxyOutput(p, generation, out); err != nil {
557- return proxyLinkStop, err
558- }
559- return proxyLinkResume, nil
560- case ports.MsgCommandResult:
561- result, err := ports.UnmarshalCommandResult(frame.Payload)
562- if err != nil {
563- return proxyLinkStop, err
564- }
565- p.completeCommandResult(generation, result)
566- return proxyLinkResume, nil
567- case ports.MsgPong:
568- _, err := ports.UnmarshalPong(frame.Payload)
569- return proxyLinkResume, err
570- case ports.MsgPing:
571- if _, err := ports.UnmarshalPing(frame.Payload); err != nil {
572- return proxyLinkStop, err
573- }
574- if err := p.sendGeneration(generation, ports.Frame{Type: ports.MsgPong, Payload: ports.MarshalPong(ports.Pong{})}); err != nil {
575- return proxyLinkStop, fmt.Errorf("proxy pong reply: %w: %w", errProxyLinkSend, err)
576- }
577- return proxyLinkResume, nil
578- case ports.MsgDetached:
579- detached, err := ports.UnmarshalDetached(frame.Payload)
580- if err != nil {
581- return proxyLinkStop, err
582- }
583- if detached.Reason == ports.ReasonReplaced {
584- return proxyLinkReplaced, nil
585- }
586- return proxyLinkStop, nil
587- case ports.MsgError:
588- remoteErr, err := ports.UnmarshalErrorMsg(frame.Payload)
589- if err != nil {
590- return proxyLinkStop, err
591- }
592- return proxyLinkStop, fmt.Errorf("remote proxy error (%d): %s", remoteErr.Code, remoteErr.Text)
593- default:
594- return proxyLinkStop, fmt.Errorf("unexpected proxy server frame type %d", frame.Type)
595- }
596-}
597-
59811 func TestProxyOutputRejectsStaleStateAndRequestsOneReset(t *testing.T) {
59912 tests := []struct {
60013 name string
@@ -1,600 +1,13 @@
1 package daemon1 package daemon
2 2
3 import (3 import (
4- "context"4+ "testing"
5- "errors"
6- "fmt"
7- "io"
8- "time"
9 5
6+ "github.com/bnema/vev/internal/domain"
10 "github.com/bnema/vev/internal/ports"7 "github.com/bnema/vev/internal/ports"
8+ "github.com/stretchr/testify/require"
11 )9 )
12 10
13-const (
14- proxyAttachedCommandTimeout = 10 * time.Second
15- proxyResumeInitialBackoff = 250 * time.Millisecond
16- proxyResumeMaxBackoff = 4 * time.Second
17- proxyResumeMaxAttempts = 5
18- proxyResumeStableDuration = 30 * time.Second
19-)
20-
21-var (
22- errProxyCommandTimeout = errors.New("proxy session: attached command timed out")
23- errProxyCommandUnavailable = errors.New("proxy session: attached command link is unavailable")
24- errProxyCommandGeneration = errors.New("proxy session: attached command link was replaced")
25- // errProxyLinkSend marks a reply that could not be written to the remote.
26- // The received frame was well formed, so the link is resumable rather than
27- // terminally broken.
28- errProxyLinkSend = errors.New("proxy session: link reply send failed")
29-)
30-
31-type proxyCommandOutcome struct {
32- result ports.CommandResult
33- err error
34-}
35-
36-type proxyCommandPending struct {
37- generation uint64
38- outcome chan proxyCommandOutcome
39-}
40-
41-type proxyRecvResult struct {
42- frame ports.Frame
43- err error
44-}
45-
46-type proxyLinkResult uint8
47-
48-const (
49- proxyLinkStop proxyLinkResult = iota
50- proxyLinkResume
51- proxyLinkReplaced
52-)
53-
54-// dialProxyHandshake installs a candidate transport, sends an ordinary Hello
55-// through the session's sole sender, and requires a matching Welcome. The
56-// candidate is never registry-visible.
57-func (d *Daemon) dialProxyHandshake(ctx context.Context, p *proxySession, intent uint8) error {
58- if err := ctx.Err(); err != nil {
59- return err
60- }
61- dialer, err := d.remoteDialerFactory.DialerForRemote(p.key.Host, p.key.Name, d.remoteTransportMode, d.log)
62- if err != nil {
63- return fmt.Errorf("proxy session: select remote dialer: %w", err)
64- }
65- if dialer == nil {
66- return errors.New("proxy session: remote dialer factory returned nil dialer")
67- }
68- if err := ctx.Err(); err != nil {
69- return err
70- }
71- transport, err := dialer.Dial(ctx)
72- if err != nil {
73- return fmt.Errorf("proxy session: dial: %w", err)
74- }
75- if transport == nil {
76- return errors.New("proxy session: remote dialer returned nil transport")
77- }
78- if err := ctx.Err(); err != nil {
79- _ = transport.Close()
80- return err
81- }
82-
83- p.mu.Lock()
84- resumeToken := p.resumeToken
85- clientID := p.clientID
86- size := p.contentSize
87- p.mu.Unlock()
88- resume := intent == ports.IntentResume
89- if !resume {
90- resumeToken = 0
91- }
92- generation, previous := p.installTransport(transport, resume)
93- if previous != nil && previous != transport {
94- _ = previous.Close()
95- }
96- failed := true
97- defer func() {
98- if failed {
99- p.retireTransport(generation, transport)
100- _ = transport.Close()
101- }
102- }()
103- hello := ports.Hello{
104- Version: ports.ProtocolVersion,
105- Intent: intent,
106- ClientID: clientID,
107- ResumeToken: resumeToken,
108- Name: p.key.Name,
109- Size: size,
110- MaxOutputInFlight: proxyOutputWindow(transport),
111- }
112- if err := p.sendGeneration(generation, ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}); err != nil {
113- return fmt.Errorf("proxy session: send hello: %w", err)
114- }
115-
116- welcome, err := recvProxyHandshake(ctx, transport, p.key.Name)
117- if err != nil {
118- return err
119- }
120- if err := ctx.Err(); err != nil {
121- return err
122- }
123-
124- p.mu.Lock()
125- if p.linkGeneration != generation || p.transport != transport {
126- p.mu.Unlock()
127- return errors.New("proxy session: link replaced during handshake")
128- }
129- p.resumeToken = welcome.ResumeToken
130- p.linkState = ports.LinkStateConnected
131- p.expired = false
132- p.mu.Unlock()
133- failed = false
134- return nil
135-}
136-
137-func proxyOutputWindow(transport ports.Transport) uint8 {
138- if _, ok := transport.(ports.DatagramTransport); ok {
139- return 1
140- }
141- return maxUnackedOutputStates
142-}
143-
144-func recvProxyHandshake(ctx context.Context, transport ports.Transport, sessionName string) (ports.Welcome, error) {
145- frame, err := recvProxyHandshakeFrame(ctx, transport)
146- if err != nil {
147- return ports.Welcome{}, fmt.Errorf("proxy session: receive welcome: %w", err)
148- }
149- welcome, err := validateProxyWelcome(frame, sessionName)
150- if err != nil {
151- return ports.Welcome{}, err
152- }
153- return welcome, nil
154-}
155-
156-func validateProxyWelcome(frame ports.Frame, sessionName string) (ports.Welcome, error) {
157- if frame.Type == ports.MsgError {
158- remoteErr, err := ports.UnmarshalErrorMsg(frame.Payload)
159- if err != nil {
160- return ports.Welcome{}, fmt.Errorf("proxy session: malformed remote error: %w", err)
161- }
162- return ports.Welcome{}, fmt.Errorf("proxy session: remote rejected handshake (%d): %s", remoteErr.Code, remoteErr.Text)
163- }
164- if frame.Type != ports.MsgWelcome {
165- return ports.Welcome{}, fmt.Errorf("proxy session: first server frame is %d, want Welcome", frame.Type)
166- }
167- welcome, err := ports.UnmarshalWelcome(frame.Payload)
168- if err != nil {
169- return ports.Welcome{}, fmt.Errorf("proxy session: decode welcome: %w", err)
170- }
171- if welcome.SessionID == "" || welcome.SessionName != sessionName {
172- return ports.Welcome{}, fmt.Errorf("proxy session: welcome identity mismatch: got %q", welcome.SessionName)
173- }
174- return welcome, nil
175-}
176-
177-func recvProxyHandshakeFrame(ctx context.Context, transport ports.Transport) (ports.Frame, error) {
178- result := make(chan proxyRecvResult, 1)
179- go func() {
180- frame, err := transport.Recv()
181- result <- proxyRecvResult{frame: frame, err: err}
182- }()
183- select {
184- case received := <-result:
185- return received.frame, received.err
186- case <-ctx.Done():
187- _ = transport.Close()
188- received := <-result // Transport.Close is required to unblock Recv.
189- if received.err != nil && !errors.Is(received.err, io.EOF) {
190- return ports.Frame{}, errors.Join(ctx.Err(), received.err)
191- }
192- return ports.Frame{}, ctx.Err()
193- }
194-}
195-
196-func (p *proxySession) installTransport(transport ports.Transport, _ bool) (uint64, ports.Transport) {
197- // Link replacement participates in the same sender serialization as every
198- // frame. A request can therefore only be published wholly before the
199- // generation changes or wholly after it.
200- p.sendMu.Lock()
201- defer p.sendMu.Unlock()
202- p.mu.Lock()
203- previous := p.transport
204- p.linkGeneration++
205- generation := p.linkGeneration
206- // Each transport must establish its ordinary output dependency with a
207- // dependency-free full frame before incremental bytes are forwarded.
208- p.resetOutputStateLocked()
209- p.transport = transport
210- // Authentication is complete after the remote Welcome; ordinary output
211- // establishes the dependency chain asynchronously.
212- p.linkState = ports.LinkStateProbing
213- p.mu.Unlock()
214- p.failPendingCommandsLocked(errProxyCommandGeneration)
215- return generation, previous
216-}
217-
218-func (p *proxySession) retireTransport(generation uint64, transport ports.Transport) {
219- p.sendMu.Lock()
220- defer p.sendMu.Unlock()
221- p.mu.Lock()
222- current := p.linkGeneration == generation && p.transport == transport
223- if current {
224- p.transport = nil
225- }
226- p.mu.Unlock()
227- if current {
228- p.failPendingCommandsLocked(errProxyCommandUnavailable)
229- }
230-}
231-
232-func (p *proxySession) sendGeneration(generation uint64, frame ports.Frame) error {
233- p.sendMu.Lock()
234- defer p.sendMu.Unlock()
235- p.mu.Lock()
236- if p.linkGeneration != generation || p.transport == nil {
237- p.mu.Unlock()
238- return errors.New("proxy session: stale or unavailable link")
239- }
240- transport := p.transport
241- p.mu.Unlock()
242- return transport.Send(frame)
243-}
244-
245-// sendCommand sends one correlated attached palette command and waits for its
246-// exact result. commandMu serializes interactive requests while sendMu remains
247-// available to all other link traffic during the wait.
248-func (p *proxySession) sendCommand(ctx context.Context, clock ports.Clock, slug string, args []string) (ports.CommandResult, error) {
249- if p == nil || clock == nil {
250- return ports.CommandResult{}, errProxyCommandUnavailable
251- }
252- if ctx == nil {
253- ctx = context.Background()
254- }
255- if err := ctx.Err(); err != nil {
256- return ports.CommandResult{}, err
257- }
258- p.commandMu.Lock()
259- defer p.commandMu.Unlock()
260- if err := ctx.Err(); err != nil {
261- return ports.CommandResult{}, err
262- }
263-
264- p.sendMu.Lock()
265- p.mu.Lock()
266- generation, transport := p.linkGeneration, p.transport
267- p.mu.Unlock()
268- if transport == nil {
269- p.sendMu.Unlock()
270- return ports.CommandResult{}, errProxyCommandUnavailable
271- }
272- p.commandNext++
273- if p.commandNext == 0 {
274- p.commandNext++
275- }
276- requestID := p.commandNext
277- payload, err := ports.MarshalCommandRequest(ports.CommandRequest{
278- Version: ports.ProtocolVersion, RequestID: requestID, Attached: true,
279- Slug: slug, Args: append([]string(nil), args...),
280- })
281- if err != nil {
282- p.sendMu.Unlock()
283- return ports.CommandResult{}, err
284- }
285- outcome := make(chan proxyCommandOutcome, 1)
286- p.commandPending[requestID] = proxyCommandPending{generation: generation, outcome: outcome}
287- if err := transport.Send(ports.Frame{Type: ports.MsgCommand, Payload: payload}); err != nil {
288- delete(p.commandPending, requestID)
289- p.sendMu.Unlock()
290- return ports.CommandResult{}, err
291- }
292- p.sendMu.Unlock()
293-
294- timer := clock.NewTimer(proxyAttachedCommandTimeout)
295- if timer == nil {
296- p.removePendingCommand(requestID, generation)
297- return ports.CommandResult{}, errProxyCommandTimeout
298- }
299- defer timer.Stop()
300- select {
301- case completed := <-outcome:
302- return completed.result, completed.err
303- case <-ctx.Done():
304- p.removePendingCommand(requestID, generation)
305- return ports.CommandResult{}, ctx.Err()
306- case <-timer.C():
307- p.removePendingCommand(requestID, generation)
308- return ports.CommandResult{}, errProxyCommandTimeout
309- }
310-}
311-
312-func (p *proxySession) removePendingCommand(requestID, generation uint64) {
313- p.sendMu.Lock()
314- pending, ok := p.commandPending[requestID]
315- if ok && pending.generation == generation {
316- delete(p.commandPending, requestID)
317- }
318- p.sendMu.Unlock()
319-}
320-
321-// completeCommandResult ignores malformed correlation state, unknown/late IDs,
322-// and results received by a replacement generation.
323-func (p *proxySession) completeCommandResult(generation uint64, result ports.CommandResult) {
324- if result.RequestID == 0 {
325- return
326- }
327- p.sendMu.Lock()
328- pending, ok := p.commandPending[result.RequestID]
329- if !ok || pending.generation != generation {
330- p.sendMu.Unlock()
331- return
332- }
333- delete(p.commandPending, result.RequestID)
334- p.sendMu.Unlock()
335- pending.outcome <- proxyCommandOutcome{result: result}
336-}
337-
338-// failPendingCommandsLocked requires sendMu and never blocks.
339-func (p *proxySession) failPendingCommandsLocked(err error) {
340- for id, pending := range p.commandPending {
341- delete(p.commandPending, id)
342- pending.outcome <- proxyCommandOutcome{err: err}
343- }
344-}
345-
346-func (d *Daemon) runProxyLink(ctx context.Context, p *proxySession) {
347- defer p.finish()
348- connectedAt := d.clock.Now()
349- resumeAttempts := 0
350- for {
351- p.mu.Lock()
352- generation := p.linkGeneration
353- transport := p.transport
354- p.mu.Unlock()
355- if transport == nil {
356- return
357- }
358-
359- result := d.runProxyTransport(ctx, p, generation, transport)
360- if result == proxyLinkReplaced {
361- // ReasonReplaced is terminal for the exact transport incarnation. Publish
362- // expiry before retirement clears the transport identity used by the
363- // callback fence; the switch below deliberately never redials it.
364- d.markProxyReplaced(p, generation, transport)
365- }
366- p.retireTransport(generation, transport)
367- switch result {
368- case proxyLinkReplaced, proxyLinkStop:
369- return
370- case proxyLinkResume:
371- if ctx.Err() != nil {
372- return
373- }
374- if d.clock.Now().Sub(connectedAt) >= proxyResumeStableDuration {
375- resumeAttempts = 0
376- }
377- d.updateProxyDisconnectedState(p, generation, ports.LinkStateOffline)
378-
379- for {
380- if resumeAttempts >= proxyResumeMaxAttempts {
381- d.log.Warn("proxy resume attempts exhausted", "host", p.key.Host, "session", p.key.Name, "attempts", resumeAttempts)
382- return
383- }
384- delay := proxyResumeBackoff(resumeAttempts)
385- if !waitProxyResume(ctx, d.clock, delay) {
386- return
387- }
388- resumeAttempts++
389- if err := d.dialProxyHandshake(ctx, p, ports.IntentResume); err != nil {
390- p.mu.Lock()
391- failedGeneration := p.linkGeneration
392- p.mu.Unlock()
393- d.updateProxyDisconnectedState(p, failedGeneration, ports.LinkStateOffline)
394- if ctx.Err() != nil {
395- return
396- }
397- d.log.Warn("proxy resume failed", "host", p.key.Host, "session", p.key.Name, "attempt", resumeAttempts, "err", err)
398- continue
399- }
400- connectedAt = d.clock.Now()
401- d.repaintProxyLifecycle(p)
402- break
403- }
404- }
405- }
406-}
407-
408-func proxyResumeBackoff(attempt int) time.Duration {
409- delay := proxyResumeInitialBackoff
410- for range attempt {
411- if delay >= proxyResumeMaxBackoff/2 {
412- return proxyResumeMaxBackoff
413- }
414- delay *= 2
415- }
416- return min(delay, proxyResumeMaxBackoff)
417-}
418-
419-func waitProxyResume(ctx context.Context, clock ports.Clock, delay time.Duration) bool {
420- if err := ctx.Err(); err != nil {
421- return false
422- }
423- timer := clock.NewTimer(delay)
424- if timer == nil {
425- return false
426- }
427- defer timer.Stop()
428- select {
429- case <-ctx.Done():
430- return false
431- case <-timer.C():
432- return true
433- }
434-}
435-
436-func (d *Daemon) runProxyTransport(ctx context.Context, p *proxySession, generation uint64, transport ports.Transport) proxyLinkResult {
437- recvCtx, cancelRecv := context.WithCancel(ctx)
438- recv := make(chan proxyRecvResult, 1)
439- recvDone := make(chan struct{})
440- go func() {
441- defer close(recvDone)
442- for {
443- frame, err := transport.Recv()
444- select {
445- case recv <- proxyRecvResult{frame: frame, err: err}:
446- case <-recvCtx.Done():
447- return
448- }
449- if err != nil {
450- return
451- }
452- }
453- }()
454- defer func() {
455- cancelRecv()
456- _ = transport.Close()
457- <-recvDone
458- }()
459-
460- var events <-chan ports.LinkEvent
461- if reporter, ok := transport.(ports.LinkStateReporter); ok {
462- events = reporter.LinkEvents()
463- }
464- for {
465- select {
466- case <-ctx.Done():
467- return proxyLinkStop
468- case event, ok := <-events:
469- if !ok {
470- events = nil
471- continue
472- }
473- if !d.updateProxyLinkState(p, generation, transport, event.State) {
474- return proxyLinkStop
475- }
476- if event.State == ports.LinkStateOffline || event.State == ports.LinkStateDead {
477- _ = transport.Close()
478- return proxyLinkResume
479- }
480- case received := <-recv:
481- if received.err != nil {
482- if ctx.Err() != nil {
483- return proxyLinkStop
484- }
485- return proxyLinkResume
486- }
487- result, err := d.handleLinkFrame(p, generation, received.frame)
488- if err != nil {
489- // A failed reply says nothing about the remote's framing, so it
490- // retires this transport instead of the whole proxy link.
491- if errors.Is(err, errProxyLinkSend) {
492- d.log.Warn("proxy link reply failed", "host", p.key.Host, "session", p.key.Name, "err", err)
493- return proxyLinkResume
494- }
495- d.log.Warn("invalid proxy link frame", "host", p.key.Host, "session", p.key.Name, "err", err)
496- return proxyLinkStop
497- }
498- if result != proxyLinkResume {
499- return result
500- }
501- }
502- }
503-}
504-
505-// handleProxySideEffect forwards a stateless remote effect to the exact local
506-// attachment that currently owns this proxy. Connection admission and transport
507-// incarnation checks make a stale handoff a harmless drop rather than a send
508-// to the client's next session.
509-func (d *Daemon) handleProxySideEffect(p *proxySession, generation uint64, out ports.Output) error {
510- if p == nil || !p.currentLinkGeneration(generation) {
511- return nil
512- }
513- attachments := snapshotAttachmentSession(p)
514- if len(attachments) == 0 {
515- return nil
516- }
517- ac := attachments[0]
518- if ac.output == nil {
519- return nil
520- }
521- expected := ac.transportSnapshot()
522- if expected.transport == nil {
523- return nil
524- }
525- token := attachmentToken(p, ac, expected.transport)
526- if token.sess != p {
527- return nil
528- }
529- ticket, admitted := ac.beginAttachmentEffect(token)
530- if !admitted {
531- return nil
532- }
533- defer ticket.End()
534- if !p.currentLinkGeneration(generation) {
535- return nil
536- }
537- frame := ac.output.sideEffect(out.Data, ac.echoAck.Load())
538- if err := ac.sendExpectedTransportForAttachment(expected, frame, ticket); err != nil {
539- if errors.Is(err, errAttachmentTransition) {
540- return nil
541- }
542- return fmt.Errorf("proxy session: side-effect send: %w: %w", errProxyLinkSend, err)
543- }
544- return nil
545-}
546-
547-// handleLinkFrame returns proxyLinkResume to mean "continue this transport";
548-// the other values are terminal actions.
549-func (d *Daemon) handleLinkFrame(p *proxySession, generation uint64, frame ports.Frame) (proxyLinkResult, error) {
550- switch frame.Type {
551- case ports.MsgOutput:
552- out, err := ports.UnmarshalOutput(frame.Payload)
553- if err != nil {
554- return proxyLinkStop, err
555- }
556- if err := d.handleProxyOutput(p, generation, out); err != nil {
557- return proxyLinkStop, err
558- }
559- return proxyLinkResume, nil
560- case ports.MsgCommandResult:
561- result, err := ports.UnmarshalCommandResult(frame.Payload)
562- if err != nil {
563- return proxyLinkStop, err
564- }
565- p.completeCommandResult(generation, result)
566- return proxyLinkResume, nil
567- case ports.MsgPong:
568- _, err := ports.UnmarshalPong(frame.Payload)
569- return proxyLinkResume, err
570- case ports.MsgPing:
571- if _, err := ports.UnmarshalPing(frame.Payload); err != nil {
572- return proxyLinkStop, err
573- }
574- if err := p.sendGeneration(generation, ports.Frame{Type: ports.MsgPong, Payload: ports.MarshalPong(ports.Pong{})}); err != nil {
575- return proxyLinkStop, fmt.Errorf("proxy pong reply: %w: %w", errProxyLinkSend, err)
576- }
577- return proxyLinkResume, nil
578- case ports.MsgDetached:
579- detached, err := ports.UnmarshalDetached(frame.Payload)
580- if err != nil {
581- return proxyLinkStop, err
582- }
583- if detached.Reason == ports.ReasonReplaced {
584- return proxyLinkReplaced, nil
585- }
586- return proxyLinkStop, nil
587- case ports.MsgError:
588- remoteErr, err := ports.UnmarshalErrorMsg(frame.Payload)
589- if err != nil {
590- return proxyLinkStop, err
591- }
592- return proxyLinkStop, fmt.Errorf("remote proxy error (%d): %s", remoteErr.Code, remoteErr.Text)
593- default:
594- return proxyLinkStop, fmt.Errorf("unexpected proxy server frame type %d", frame.Type)
595- }
596-}
597-
598 func TestProxyOutputRejectsStaleStateAndRequestsOneReset(t *testing.T) {11 func TestProxyOutputRejectsStaleStateAndRequestsOneReset(t *testing.T) {
599 tests := []struct {12 tests := []struct {
600 name string13 name string
M internal/usecase/daemon/session_test.go +1 -2
@@ -171,7 +171,6 @@ func TestCreateTabForAttachmentUsesRequestingAttachmentViewportAndTheme(t *testi
171171 Intent: ports.IntentAttach,
172172 Name: "work",
173173 ClientID: [16]byte{2},
174- Proxied: true,
175174 Size: domain.Size{Cols: 80, Rows: 24},
176175 }, secondTransport)
177176 require.NoError(t, err)
@@ -192,7 +191,7 @@ func TestCreateTabForAttachmentUsesRequestingAttachmentViewportAndTheme(t *testi
192191 gotSize := tb.size
193192 pane := tb.focusedPane()
194193 tb.mu.Unlock()
195- require.Equal(t, domain.Size{Cols: 80, Rows: 24}, gotSize, "proxied sizing must come from the requesting attachment")
194+ require.Equal(t, contentSize(domain.Size{Cols: 80, Rows: 24}), gotSize, "sizing must come from the requesting attachment")
196195 assertPaneDefaultColors(t, pane, requesterTheme.Foreground, requesterTheme.Background)
197196
198197 require.NoError(t, d.killSession(sess, ports.ReasonSessionKilled, false))
@@ -171,7 +171,6 @@ func TestCreateTabForAttachmentUsesRequestingAttachmentViewportAndTheme(t *testi
171 Intent: ports.IntentAttach,171 Intent: ports.IntentAttach,
172 Name: "work",172 Name: "work",
173 ClientID: [16]byte{2},173 ClientID: [16]byte{2},
174- Proxied: true,
175 Size: domain.Size{Cols: 80, Rows: 24},174 Size: domain.Size{Cols: 80, Rows: 24},
176 }, secondTransport)175 }, secondTransport)
177 require.NoError(t, err)176 require.NoError(t, err)
@@ -192,7 +191,7 @@ func TestCreateTabForAttachmentUsesRequestingAttachmentViewportAndTheme(t *testi
192 gotSize := tb.size191 gotSize := tb.size
193 pane := tb.focusedPane()192 pane := tb.focusedPane()
194 tb.mu.Unlock()193 tb.mu.Unlock()
195- require.Equal(t, domain.Size{Cols: 80, Rows: 24}, gotSize, "proxied sizing must come from the requesting attachment")194+ require.Equal(t, contentSize(domain.Size{Cols: 80, Rows: 24}), gotSize, "sizing must come from the requesting attachment")
196 assertPaneDefaultColors(t, pane, requesterTheme.Foreground, requesterTheme.Background)195 assertPaneDefaultColors(t, pane, requesterTheme.Foreground, requesterTheme.Background)
197 196
198 require.NoError(t, d.killSession(sess, ports.ReasonSessionKilled, false))197 require.NoError(t, d.killSession(sess, ports.ReasonSessionKilled, false))

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